code: def _flow_lines(self):
      tags = []
      for step in self.steps:
          tags.extend(tag.split("(")[0] for tag in step.tags)
  
      yield 0, "# -*- coding: utf-8 -*-"
      yield 0, "from metaflow import FlowSpec, step, Parameter, project, IncludeFile, JSONType, current, parallel"
      yield 0, "from metaflow_test import assert_equals, assert_equals_metadata, assert_exception, ExpectationFailed, is_resumed, ResumeFromHere, TestRetry, try_to_get_card"
      if tags:
          yield 0, "from metaflow import %s" % ",".join(tags)
  
      yield 0, self.test.HEADER
      yield 0, "class %s(FlowSpec):" % self.flow_name
  
      for var, val in self.test.CLASS_VARS.items():
          yield 1, "%s = %s" % (var, val)
  
      for var, parameter in self.test.PARAMETERS.items():
          kwargs = ["%s=%s" % (k, v) for k, v in parameter.items()]
          yield 1, '%s = Parameter("%s", %s)' % (
              var,
              var,
              ",".join(kwargs),
          )
  
      for var, include in self.test.INCLUDE_FILES.items():
          kwargs = ["%s=%s" % (k, v) for k, v in include.items()]
          yield 1, '%s = IncludeFile("%s", %s)' % (
              var,
              var,
              ",".join(kwargs),
          )
  
      for name, node in self.graphspec["graph"].items():
          step = self._choose_step(name, node)
          self.used.add(step)
  
          for tagspec in step.tags:
              yield 1, "@%s" % tagspec
  
          if "parallel_step" in node:
              yield 1, "@parallel"
  
          yield 1, "@step"
  
          if "join" in node:
              yield 1, "def %s(self, inputs):" % name
          else:
              yield 1, "def %s(self):" % name
  
          if "foreach" in node:
              yield 2, "self.%s = %s" % (
                  node["foreach_var"],
                  node["foreach_var_default"],
              )
  
          for line in self._format_method(step):
              yield 2, line
  
          if "linear" in node:
              yield 2, "self.next(self.%s)" % node["linear"]
          elif "branch" in node:
              branches = ",".join("self.%s" % x for x in node["branch"])
              yield 2, "self.next(%s)" % branches
          elif "foreach" in node:
              yield 2, 'self.next(self.%s, foreach="%s")' % (
                  node["foreach"],
                  node["foreach_var"],
              )
          elif "num_parallel" in node:
              yield 2, "self.next(self.%s, num_parallel=%d)" % (
                  node["parallel"],
                  node["num_parallel"],
              )
  
      yield 0, "if __name__ == '__main__':"
      yield 1, "%s()" % self.flow_name
  
file_path: metaflow/test/core/metaflow_test/formatter.py
imports: import sys
  import inspect
label_desc: Use image node_id 6 for calling the FlowFormatter obj's underlying member method code with example usage: obj._flow_lines() without return types