Write all the nodes in the workflow to the DAX file.

def write_abstract_dag(self):
    """
    Write all the nodes in the workflow to the DAX file.
    """

    # keep track of if we are using stampede at TACC
    using_stampede = False

    if not self.__dax_file_path:
      # this workflow is not dax-compatible, so don't write a dax
      return

    import Pegasus.DAX3
    # create the workflow object
    dax_name = os.path.split(self.__dax_file_path)[-1]
    dax_basename = '.'.join(dax_name.split('.')[0:-1])
    workflow = Pegasus.DAX3.ADAG( dax_basename )

    # we save the ID number to DAG node name mapping so that
    # we can easily write out the child/parent relationship
    # later
    node_job_object_dict = {}

    # FIXME disctionary of executables and pfns in the workflow
    # Pegasus should take care of this so we don't have to
    workflow_executable_dict = {}
    workflow_pfn_dict = {}

    # Add PFN caches for this workflow
    for pfn_tuple in self.get_pfn_cache():
        workflow_pfn_dict[pfn_tuple[0]] = pfn_tuple

    if self.get_pegasus_worker():
      # write the executable into the dax
      worker_package = Pegasus.DAX3.Executable(
        namespace="pegasus", name="worker",
        os="linux", arch="x86_64", installed=False)
      worker_package.addPFN(Pegasus.DAX3.PFN(self.get_pegasus_worker(),"local"))
      workflow_executable_dict['pegasus-pegasus_worker'] = worker_package

    # check for the pegasus-cluster package
    for path in os.environ["PATH"].split(":"):
      cluster_path = os.path.join(path,"pegasus-cluster")
      if os.path.exists(cluster_path):
        # and add to the dax if it exists
        seqexec_package = Pegasus.DAX3.Executable(
          namespace="pegasus", name="seqexec",
          os="linux", arch="x86_64", installed=True)
        seqexec_package.addPFN(Pegasus.DAX3.PFN(cluster_path,"local"))
        workflow_executable_dict['pegasus-pegasus_seqexec'] = seqexec_package

    id = 0
    for node in self.__nodes:
      if self.is_dax() and isinstance(node, LSCDataFindNode):
        pass

      elif isinstance(node.job(), CondorDAGManJob):
        id += 1
        id_tag = "ID%06d" % id
        node_name = node._CondorDAGNode__name

        if node.job().get_dax() is None:
          # write this node as a sub-dag
          subdag_name = os.path.split(node.job().get_dag())[-1]
          try:
            subdag_exec_path = os.path.join(
              os.getcwd(),node.job().get_dag_directory())
          except AttributeError:
            subdag_exec_path = os.getcwd()

          subdag = Pegasus.DAX3.DAG(subdag_name,id=id_tag)
          subdag.addProfile(Pegasus.DAX3.Profile("dagman","DIR",subdag_exec_path))

          subdag_file = Pegasus.DAX3.File(subdag_name)
          subdag_file.addPFN(Pegasus.DAX3.PFN(os.path.join(subdag_exec_path,subdag_name),"local"))
          workflow.addFile(subdag_file)
          workflow.addDAG(subdag)
          node_job_object_dict[node_name] = subdag

        else:
          # write this node as a sub-dax
          subdax_name = os.path.split(node.job().get_dax())[-1]
          dax_subdir = node.job().get_dag_directory()
          if dax_subdir:
            subdax_path = os.path.join(
              os.getcwd(),node.job().get_dag_directory(),subdax_name)
          else:
            subdax_path = os.path.join(os.getcwd(),subdax_name)
            dax_subdir = '.'

          subdax = Pegasus.DAX3.DAX(subdax_name,id=id_tag)

          # FIXME pegasus should ensure these are unique
          for pfn_tuple in node.job().get_pfn_cache():
            workflow_pfn_dict[pfn_tuple[0]] = pfn_tuple

          # set the storage, execute, and output directory locations
          pegasus_args = """--dir %s """ % dax_subdir
          pegasus_args += """--output-dir %s """ % dax_subdir

          # set the maxjobs categories for the subdax
          # FIXME pegasus should expose this in the dax, so it can
          # be handled like the MAXJOBS keyword in dag files
          for maxjobcat in node.get_maxjobs_categories():
            pegasus_args += "-Dpegasus.dagman." + maxjobcat[0] + ".maxjobs=" + maxjobcat[1] + " "

          if not self.is_dax():
            pegasus_args += "--nocleanup "

          if node.get_cluster_jobs():
            pegasus_args += "--cluster " + node.get_cluster_jobs() + " "

          if node.get_reduce_dax() is False:
            pegasus_args += " --force "

          if node.get_static_pfn_cache():
            pegasus_args += " --cache " + node.get_static_pfn_cache() + " "

          pegasus_args += "--output-site local -vvvvvv"
          subdax.addArguments(pegasus_args)

          subdax_file = Pegasus.DAX3.File(subdax_name)
          subdax_file.addPFN(Pegasus.DAX3.PFN(subdax_path,"local"))
          workflow.addFile(subdax_file)
          workflow.addDAX(subdax)
          node_job_object_dict[node_name] = subdax

      else:
        # write this job as a regular node
        executable = node.job()._CondorJob__executable
        node_name = node._CondorDAGNode__name

        id += 1
        id_tag = "ID%06d" % id
        node_job_object_dict[node_name] = id_tag

        # get the name of the executable
        executable_namespace = 'ligo-' + str(node.job().__class__.__name__).lower()
        executable_base = os.path.basename(executable)

        workflow_job = Pegasus.DAX3.Job( namespace=executable_namespace,
          name=executable_base, version="1.0", id=id_tag)

        cmd_line = node.get_cmd_tuple_list()

        # loop through all filenames looking for them in the command
        # line so that they can be replaced appropriately by xml tags
        input_node_file_dict = {}
        for f in node.get_input_files():
          input_node_file_dict[f] = 1

        for f in input_node_file_dict.keys():
          workflow_job.uses(Pegasus.DAX3.File(os.path.basename(f)),link=Pegasus.DAX3.Link.INPUT,register=False,transfer=True)

        output_node_file_dict = {}
        for f in node.get_output_files():
          output_node_file_dict[f] = 1

        checkpoint_node_file_dict = {}
        for f in node.get_checkpoint_files():
            checkpoint_node_file_dict[f] = 1

        for f in output_node_file_dict.keys():
          workflow_job.uses(Pegasus.DAX3.File(os.path.basename(f)),link=Pegasus.DAX3.Link.OUTPUT,register=False,transfer=True)

        for f in checkpoint_node_file_dict.keys():
          workflow_job.uses(Pegasus.DAX3.File(os.path.basename(f)),link=Pegasus.DAX3.Link.CHECKPOINT,register=False,transfer=True)

        node_file_dict = dict( input_node_file_dict.items() + output_node_file_dict.items() + checkpoint_node_file_dict.items() )

        for job_arg in cmd_line:
          try:
            if node_file_dict.has_key(job_arg[0]):
              workflow_job.addArguments(Pegasus.DAX3.File(os.path.basename(job_arg[0])))
            elif node_file_dict.has_key(job_arg[1]):
              workflow_job.addArguments(job_arg[0], Pegasus.DAX3.File(os.path.basename(job_arg[1])))
            elif len(job_arg[1].split(' ')) != 1:
                args = [job_arg[0]]
                for arg in job_arg[1].split(' '):
                    if node_file_dict.has_key(arg):
                        args.append(Pegasus.DAX3.File(os.path.basename(arg)))
                    else:
                        args.append(arg)
                workflow_job.addArguments(*args)
            else:
              workflow_job.addArguments(job_arg[0], job_arg[1])
          except IndexError:
            pass

        # Check for desired grid site
        if node.job().get_grid_site():
            this_grid_site = node.job().get_grid_site()
            workflow_job.addProfile(Pegasus.DAX3.Profile('hints','execution.site',this_grid_site))
            if this_grid_site == 'stampede-dev' or this_grid_site=='stampede':
              using_stampede = True

        # write the executable into the dax
        job_executable = Pegasus.DAX3.Executable(
          namespace=executable_namespace,
          name=executable_base, version="1.0",
          os="linux", arch="x86_64",
          installed=node.job().get_executable_installed())

        executable_path = os.path.join(os.getcwd(),executable)
        job_executable.addPFN(Pegasus.DAX3.PFN(executable_path,"local"))

        workflow_executable_dict[executable_namespace + executable_base] = job_executable

        # write the mpi cluster parameter for the job
        if node.job().get_dax_mpi_cluster():
          workflow_job.addProfile(Pegasus.DAX3.Profile("pegasus","job.aggregator","mpiexec"))
          workflow_job.addProfile(Pegasus.DAX3.Profile("pegasus","clusters.size",str(node.job().get_dax_mpi_cluster())))

        # write the grid start parameter for this node
        # if the grid start is not None
        if node.get_grid_start():
          workflow_job.addProfile(Pegasus.DAX3.Profile("pegasus","gridstart",node.get_grid_start()))

        # write the bundle parameter if this node has one
        if node.get_dax_collapse():
          workflow_job.addProfile(Pegasus.DAX3.Profile("pegasus","clusters.size",str(node.get_dax_collapse())))

        # write number of times the node should be retried
        if node.get_retry():
          workflow_job.addProfile(Pegasus.DAX3.Profile("dagman","retry",str(node.get_retry())))

        # write the post script for this node
        if node.get_post_script():
          post_script_base = os.path.basename(node.get_post_script())
          post_script_path = os.path.join(os.getcwd(),node.get_post_script())
          workflow_job.addProfile(Pegasus.DAX3.Profile("dagman","post",post_script_base))
          workflow_job.addProfile(Pegasus.DAX3.Profile("dagman","post.path." + post_script_base,post_script_path))

        # write the post script for this node
        if node.get_post_script_arg():
          workflow_job.addProfile(Pegasus.DAX3.Profile("dagman","post.arguments",' '.join(node.get_post_script_arg())))

        # write the dag node category if this node has one
        if node.get_category():
          workflow_job.addProfile(Pegasus.DAX3.Profile("dagman","category",str(node.get_category())))

        # write the dag node priority if this node has one
        if node.get_priority():
          workflow_job.addProfile(Pegasus.DAX3.Profile("condor","priority",str(node.get_priority())))

        # write the universe that this job should run in to the dax
        if node.get_dax_collapse():
          # collapsed jobs must run in the vanilla universe
          workflow_job.addProfile(Pegasus.DAX3.Profile("condor","universe","vanilla"))
        else:
          workflow_job.addProfile(Pegasus.DAX3.Profile("condor","universe",node.job().get_universe()))

        # add any other user specified condor commands or classads
        for p in node.get_pegasus_profile():
            workflow_job.addProfile(Pegasus.DAX3.Profile(p[0],p[1],p[2]))

        # finally add this job to the workflow
        workflow.addJob(workflow_job)
        node_job_object_dict[node_name] = workflow_job


    # print parent-child relationships to DAX
    for node in self.__nodes:
      if self.is_dax() and isinstance(node, LSCDataFindNode):
        pass
      elif self.is_dax() and ( len(node._CondorDAGNode__parents) == 1 ) and isinstance(node._CondorDAGNode__parents[0], LSCDataFindNode):
        pass
      else:
        child_job_object = node_job_object_dict[str(node)]
        if node._CondorDAGNode__parents:
          for parent in node._CondorDAGNode__parents:
            if self.is_dax() and isinstance(parent, LSCDataFindNode):
              pass
            else:
              parent_job_object = node_job_object_dict[str(parent)]
              workflow.addDependency(Pegasus.DAX3.Dependency(parent=parent_job_object, child=child_job_object))

    # FIXME put all the executables in the workflow
    for exec_key in workflow_executable_dict.keys():
      workflow.addExecutable(workflow_executable_dict[exec_key])

    # FIXME if we are running on stampede, add the mpi wrapper job
    if using_stampede:
      prod_mpiexec = Pegasus.DAX3.Executable(namespace="pegasus",
        name="mpiexec", os="linux", arch="x86_64", installed="true")
      prod_mpiexec.addPFN(Pegasus.DAX3.PFN("file:///home1/02796/dabrown/bin/mpi-cluster-wrapper-impi.sh","stampede"))
      workflow.addExecutable(prod_mpiexec)

      dev_mpiexec = Pegasus.DAX3.Executable(namespace="pegasus",
        name="mpiexec", os="linux", arch="x86_64", installed="true")
      dev_mpiexec.addPFN(Pegasus.DAX3.PFN("file:///home1/02796/dabrown/bin/mpi-cluster-wrapper-impi.sh","stampede-dev"))
      workflow.addExecutable(dev_mpiexec)

    # FIXME put all the pfns in the workflow
    for pfn_key in workflow_pfn_dict.keys():
      f = Pegasus.DAX3.File(workflow_pfn_dict[pfn_key][0])
      f.addPFN(Pegasus.DAX3.PFN(workflow_pfn_dict[pfn_key][1],workflow_pfn_dict[pfn_key][2]))
      workflow.addFile(f)

    f = open(self.__dax_file_path,"w")
    workflow.writeXML(f)
    f.close()