#! /usr/bin/env python
#
#  GOAL
#    Simple pipeline: just ILE+fitting iteration jobs
#    User must provide arglists for both ILE and fitting/iteration jobs
#    Assumes user has done all necessary setup (e.g., PSDs, data selection, picking channel names, etc)
#    Will add in fit assessment jobs later
#
#  WHY DO THIS?
#    We're often reanalyzing a single stretch of data again and again. (Or otherwise have some simple setup).
#    This saves time for large-scale injection studies or systematics reruns
#
#    Intend to replace naive iteration code with reproducible, extendable/scalable workflow
#
#  SEE ALSO
#     - original shell-script-based generation scripts
#     - create_event_parameter_pipeline  # long term plan
#     - create_postprocessing_event_dag.py
#     - create_event_dag_via_grid   # basic iteration code
#
# WORKFLOW STRUCTURE
#    - Required elements:
#          cache file  (specified in  --cache <cache> in cip_args)
#          PSD files (.... --psd  in cip_args)
#          channel names
#          event time
#    - Inputs
#         args_cip.txt   # no --fname, --fname-output-...  (overrridden)
#         args_ile.txt    # no --sim-xml,  --event (overrridden)      ... though you may want to add them for command-single.sh to run properly
#    - Workspaces
#          Creates one directory for each task, for each iteration
#          Main directory:
#              - output-ILE-<iteration>.xml.gz, starting with iteration 0 (=input)
#              - *.composite files, from each consolidation step
#              - all.net : union of all *.composite files 
#   
# EXAMPLES
#    python create_event_parameter_pipeline_BasicIteration.py --ile-args args_ile.txt --cip-args args.txt 
#
# TEST SEQUENCE
#        unixhome/Projects/LIGO-ILE-Applications/communications/20180806-Me-PipelineDevelopment
#    * Trivial iteration, fitting a constant likelihood at each step: 
#           rm -rf test_workflow_trivial; make test_workflow_trivial
#         


###
###  PLANS FOR IMPROVEMENT
###
#
# * Stress-testing final fit
#       - single-iteration runs which change lnL and/or coordinates
#       - note this can be done with a DRIVER, does not need to be hardcoded?
# * BCR/BCI (single-ifo vs multi-ifo run)
#       - runs which do PE on single IFOs and 
#       - again, this can be done with a DRIVER

import argparse
import sys
import os
import shutil
import numpy as np
import RIFT.lalsimutils as lalsimutils
import lalsimulation as lalsim
import lal
import functools
import itertools
import json

# Backend-neutral pipeline namespace (htcondor/glue/slurm) provided by dag_utils_generic
from RIFT.misc.dag_utils_generic import pipeline

import RIFT.misc.dag_utils_generic as dag_utils
from RIFT.misc.dag_utils_generic import mkdir
from RIFT.misc.dag_utils_generic import which


def add_batch_ILE_nodes_to_dag(my_dag,my_ile_job,my_parent_node, my_child_node, n_max, n_group, it, n_retries=3,it_start=0,convert_psd_node_list=[],node_list_dict={},extra_parent_nodes=None):
    if extra_parent_nodes is None:
        extra_parent_nodes = []
    for event in np.arange(n_max): 
        ile_node = pipeline.CondorDAGNode(my_ile_job)
        ile_node.set_retry(n_retries)
        ile_node.add_macro("macroevent", event*n_group)
        ile_node.add_macro("macrongroup", n_group)
        ile_node.add_macro("macroiteration", it)
        ile_node.add_macro("macroiterationprev", it-1)   # Option C cal-pilot seed path
        if not(my_parent_node is None):
            ile_node.add_parent(my_parent_node)
        for node in extra_parent_nodes:
            if not(node is None):
                ile_node.add_parent(node)
        if it == it_start:
            for node in convert_psd_node_list:  # for every PSD conversion job, make sure PSD is present before we run the first iteration!
                ile_node.add_parent(node)
        if my_child_node:
            my_child_node.add_parent(ile_node) # consolidate depends on all of the individual jobs
        my_dag.add_node(ile_node)
        # global vraiable edit
        if node_list_dict:
            if it in node_list_dict:
                node_list_dict[it].append(ile_node)


def parse_ile_args_flexible(my_arg_str):
    """
    Narrower-scope parser with flexible arguments, to target
    """
    import argparse
    import shlex
    parser = argparse.ArgumentParser()
    parser.add_argument("--n-eff",type=int)
    parser.add_argument("--cache-file",type=str)
    parser.add_argument('other', nargs=argparse.REMAINDER)
    args  = parser.parse_args(shlex.split(my_arg_str))
    return args

def parse_ile_args_lazy(my_arg_str,key_target):
    str_split = my_arg_str.split('--')
    matches = [x for x in str_split if x.startswith(key_target)]
    print(matches)
    return matches[-1].split(' ')[1]

                
def parse_ile_args_for_bw(my_arg_str):
    """
    this is HARD TO MAINTAIN since it needs to match ILE exactly
    """
    # Copied verbatim from ILE
    from optparse import OptionParser, OptionGroup
    optp = OptionParser()
    optp.add_option("-c", "--cache-file", default=None, help="LIGO cache file containing all data needed.")
    optp.add_option("-C", "--channel-name", action="append", help="instrument=channel-name, e.g. H1=FAKE-STRAIN. Can be given multiple times for different instruments.")
    optp.add_option("-p", "--psd-file", action="append", help="instrument=psd-file, e.g. H1=H1_PSD.xml.gz. Can be given multiple times for different instruments.")
    optp.add_option("-k", "--skymap-file", help="Use skymap stored in given FITS file.")
    optp.add_option("-x", "--coinc-xml", help="gstlal_inspiral XML file containing coincidence information.")
    optp.add_option("-I", "--sim-xml", help="XML file containing mass grid to be evaluated")
    optp.add_option("-E", "--event", default=0,type=int, help="Event number used for this run")
    optp.add_option("--n-events-to-analyze", default=1,type=int, help="Number of events to analyze from this XML")
    optp.add_option("--soft-fail-event-range",action='store_true',help='Soft failure (exit 0) if event ID is out of range. This happens in pipelines, if we have pre-built a DAG attempting to analyze more points than we really have')
    optp.add_option("-f", "--reference-freq", type=float, default=100.0, help="Waveform reference frequency. Required, default is 100 Hz.")
    optp.add_option("--fmin-template", dest='fmin_template', type=float, default=40, help="Waveform starting frequency.  Default is 40 Hz. Also equal to starting frequency for integration") 
    optp.add_option("--fmin-template-correct-for-lmax",action='store_true',help="Modify amount of data selected, waveform starting frequency to account for l-max, to better insure all requested modes start within the targeted band")
    optp.add_option("--fmin-ifo", action='append' , help="Minimum frequency for each IFO. Implemented by setting the PSD=0 below this cutoff. Use with care.") 
    optp.add_option('--nr-group', default=None,help="If using a *ssingle specific simulation* specified on the command line, provide it here")
    optp.add_option('--nr-param', default=None,help="If using a *ssingle specific simulation* specified on the command line, provide it here")
    optp.add_option("--nr-lookup",action='store_true', help=" Look up parameters from an NR catalog, instead of using the approximant specified")
    optp.add_option("--nr-lookup-group",action='append', help="Restriction on 'group' for NR lookup")
    optp.add_option("--nr-hybrid-use",action='store_true',help="Enable use of NR (or ROM!) hybrid, using --approx as the default approximant and with a frequency fmin")
    optp.add_option("--nr-hybrid-method",default="taper_add",help="Hybridization method for NR (or ROM!).  Passed through to LALHybrid. pseudo_aligned_from22 will provide ad-hoc higher modes, if the early-time hybridization model only includes the 22 mode")
    optp.add_option("--rom-group",default=None)
    optp.add_option("--rom-param",default=None)
    optp.add_option("--rom-use-basis",default=False,action='store_true',help="Use the ROM basis for inner products.")
    optp.add_option("--rom-limit-basis-size-to",default=None,type=int)
    optp.add_option("--rom-integrate-intrinsic",default=False,action='store_true',help='Integrate over intrinsic variables. REQUIRES rom_use_basis at present. ONLY integrates in mass ratio as present')
    optp.add_option("--nr-perturbative-extraction",default=False,action='store_true')
    optp.add_option("--nr-perturbative-extraction-full",default=False,action='store_true')
    optp.add_option("--nr-use-provided-strain",default=False,action='store_true')
    optp.add_option("--no-memory",default=False,action='store_true', help="At present, turns off m=0 modes. Use with EXTREME caution only if requested by model developer")
    optp.add_option("--restricted-mode-list-file",default=None,help="A list of ALL modes to use in likelihood. Incredibly dangerous. Only use when comparing with models which provide restricted mode sets, or otherwise to isolate the effect of subsets of modes on the whole")
    optp.add_option("--use-external-EOB",default=False,action='store_true')
    optp.add_option("--maximize-only",default=False, action='store_true',help="After integrating, attempts to find the single best fitting point")
    optp.add_option("--dump-lnL-time-series",default=False, action='store_true',help="(requires --sim-xml) Dump lnL(t) at the injected parameters")
    optp.add_option("-a", "--approximant", default="TaylorT4", help="Waveform family to use for templates. Any approximant implemented in LALSimulation is valid.")
    optp.add_option("-A", "--amp-order", type=int, default=0, help="Include amplitude corrections in template waveforms up to this e.g. (e.g. 5 <==> 2.5PN), default is Newtonian order.")
    optp.add_option("--l-max", type=int, default=2, help="Include all (l,m) modes with l less than or equal to this value.")
    optp.add_option("-s", "--data-start-time", type=float, default=None, help="GPS start time of data segment. If given, must also give --data-end-time. If not given, sane start and end time will automatically be chosen.")
    optp.add_option("-e", "--data-end-time", type=float, default=None, help="GPS end time of data segment. If given, must also give --data-start-time. If not given, sane start and end time will automatically be chosen.")
    optp.add_option("--data-integration-window-half",default=50*1e-3,type=float,help="Only change this window size if you are an expert. The window for time integration is -/+ this quantity around the event time")
    optp.add_option("-F", "--fmax", type=float, help="Upper frequency of signal integration. Default is use PSD's maximum frequency.")
    optp.add_option("--srate",default=16384,type=int,help="Sampling rate. Change ONLY IF YOU ARE ABSOLUTELY SURE YOU KNOW WHAT YOU ARE DOING.")
    optp.add_option("-t", "--event-time", type=float, help="GPS time of the event --- probably the end time. Required if --coinc-xml not given.")
    optp.add_option("-i", "--inv-spec-trunc-time", type=float, default=8., help="Timescale of inverse spectrum truncation in seconds (Default is 8 - give 0 for no truncation)")
    optp.add_option("-w", "--window-shape", type=float, default=0, help="Shape of Tukey window to apply to data (default is no windowing)")
    optp.add_option("--psd-window-shape", type=float, default=0, help="Shape of Tukey window that *was* applied to the PSD being passed. If nonzero, we will rescale the PSD by the ratio of window shape results")
    optp.add_option("-m", "--time-marginalization", action="store_true", help="Perform marginalization over time via direct numerical integration. Default is false.")
    optp.add_option("-d", "--distance-marginalization", action="store_true", help="Perform marginalization over distance via a look-up table. Default is false.")
    optp.add_option("-l", "--distance-marginalization-lookup-table", default=None, help="Look-up table for distance marginalization.")
    optp.add_option("--vectorized", action="store_true", help="Perform manipulations of lm and timeseries using numpy arrays, not LAL data structures.  (Combine with --gpu to enable GPU use, where available)")
    optp.add_option("--gpu", action="store_true", help="Perform manipulations of lm and timeseries using numpy arrays, CONVERTING TO GPU when available. You MUST use this option with --vectorized (otherwise it is a no-op). You MUST have a suitable version of cupy installed, your cuda operational, etc")
    optp.add_option("--force-gpu-only", action="store_true", help="Hard fail if no GPU present (assessed by cupy not loading)")
    optp.add_option("--force-xpy", action="store_true", help="Use the xpy code path.  Use with --vectorized --gpu to use the fallback CPU-based code path. Useful for debugging.")
    optp.add_option("-o", "--output-file", help="Save result to this file.")
    optp.add_option("-O", "--output-format", default='xml', help="[xml|hdf5]")
    optp.add_option("-S", "--save-samples", action="store_true", help="Save sample points to output-file. Requires --output-file to be defined.")
    optp.add_option("-L", "--save-deltalnL", type=float, default=float("Inf"), help="Threshold on deltalnL for points preserved in output file.  Requires --output-file to be defined")
    optp.add_option("-P", "--save-P", type=float,default=0.1, help="Threshold on cumulative probability for points preserved in output file.  Requires --output-file to be defined")
    optp.add_option("--verbose",action='store_true')
    integration_params = OptionGroup(optp, "Integration Parameters", "Control the integration with these options.")
    integration_params.add_option("--n-max", type=int, help="Total number of samples points to draw. If this number is hit before n_eff, then the integration will terminate. Default is 'infinite'.",default=1e7)
    integration_params.add_option("--n-eff", type=int, default=100, help="Total number of effective samples points to calculate before the integration will terminate. Default is 100")
    integration_params.add_option("--n-chunk", type=int, help="Chunk'.",default=10000)
    integration_params.add_option("--convergence-tests-on",default=False,action='store_true')
    integration_params.add_option("--seed", type=int, help="Random seed to use. Default is to not seed the RNG.")
    integration_params.add_option("--no-adapt", action="store_true", help="Turn off adaptive sampling. Adaptive sampling is on by default.")
    integration_params.add_option("--no-adapt-distance", action="store_true", help="Turn off adaptive sampling, just for distance. Adaptive sampling is on by default.")
    integration_params.add_option("--no-adapt-after-first",action='store_true',help="Disables adaptation after first iteration with significant lnL")
    integration_params.add_option("--adapt-weight-exponent", type=float, default=1.0, help="Exponent to use with weights (likelihood integrand) when doing adaptive sampling. Used in tandem with --adapt-floor-level to prevent overconvergence. Default is 1.0.")
    integration_params.add_option("--adapt-floor-level", type=float, default=0.1, help="Floor to use with weights (likelihood integrand) when doing adaptive sampling. This is necessary to ensure the *sampling* prior is non zero during adaptive sampling and to prevent overconvergence. Default is 0.1 (no floor)")
    integration_params.add_option("--adapt-adapt",action='store_true',help="Adapt the tempering exponent")
    integration_params.add_option("--adapt-log",action='store_true',help="Use a logarithmic tempering exponent")
    integration_params.add_option("--interpolate-time", default=False,help="If using time marginalization, compute using a continuously-interpolated array. (Default=false)")
    integration_params.add_option("--d-prior",default='Euclidean' ,type=str,help="Distance prior for dL.  Options are dL^2 (Euclidean) and 'pseudo-cosmo'  .")
    integration_params.add_option("--d-max", default=10000,type=float,help="Maximum distance in volume integral. Used to SET THE PRIOR; changing this value changes the numerical answer.")
    integration_params.add_option("--d-min", default=1,type=float,help="Minimum distance in volume integral. Used to SET THE PRIOR; changing this value changes the numerical answer.")
    integration_params.add_option("--declination-cosine-sampler",action='store_true',help="If specified, the parameter used for declination is cos(dec), not dec")
    integration_params.add_option("--inclination-cosine-sampler",action='store_true',help="If specified, the parameter used for inclination is cos(dec), not dec")
    integration_params.add_option("--internal-rotate-phase", action='store_true',help="If specified, the integration sampler uses phase_p ==phi+psi and phase_m == phi-psi as sampling coordinates, both ranging from 0 to 4 pi.  The prior is twice as large.")
    integration_params.add_option("--internal-sky-network-coordinates",action='store_true',help="If specified, perform integration in sky coordinates aligned with the first two IFOs provided")
    integration_params.add_option("--manual-logarithm-offset",type=float,default=0,help="Target value of logarithm lnL. Integrand is reduced by exp(-manual_logarithm_offset).  Important for high-SNR sources!   Should be set dynamically")
    integration_params.add_option("--sampler-method",default="adaptive_cartesian",help="adaptive_cartesian|GMM|adaptive_cartesian_gpu")
    integration_params.add_option("--sampler-xpy",default=None,help="numpy|cupy  if the adaptive_cartesian_gpu sampler is active, use that.")
    optp.add_option_group(integration_params)

    intrinsic_params = OptionGroup(optp, "Intrinsic Parameters", "Intrinsic parameters (e.g component mass) to use.")
#intrinsic_params.add_option("--pin-distance-to-sim",action='store_true', help="Pin *distance* value to sim entry. Used to enable source frame reconstruction with NR.")
    intrinsic_params.add_option("--mass1", type=float, help="Value of first component mass, in solar masses. Required if not providing coinc tables.")
    intrinsic_params.add_option("--mass2", type=float, help="Value of second component mass, in solar masses. Required if not providing coinc tables.")
    intrinsic_params.add_option("--eff-lambda", type=float, help="Value of effective tidal parameter. Optional, ignored if not given.")
    intrinsic_params.add_option("--deff-lambda", type=float, help="Value of second effective tidal parameter. Optional, ignored if not given")
    optp.add_option_group(intrinsic_params)

    intrinsic_int_params = OptionGroup(optp, "Intrinsic integrated parameters", "Intrinsic parameters to integrate over. ONLY currently used with ROM version")
    intrinsic_int_params.add_option("--parameter",action='append')
    intrinsic_int_params.add_option("--parameter-range",action='append',type=str)
    intrinsic_int_params.add_option("--adapt-intrinsic",action='store_true')
    optp.add_option_group(intrinsic_int_params)


    opts, args = optp.parse_args(my_arg_str.split())  # parse full set, assume initial argument already stripped

    return opts

parser = argparse.ArgumentParser()
parser.add_argument("--working-directory",default="./")
parser.add_argument("--input-grid",default="overlap-grid.xml.gz")
parser.add_argument("--cip-args",default=None,help="filename of args_cip.txt file  which holds CIP arguments.  Should NOT conflict with arguments auto-set by this DAG ... in particular, i/o arguments will be modified. ")
parser.add_argument("--cip-args-list",default=None,help="filename of args_cip_list.file, which holds CIP arguments.  Overrides cip-args if present. One CIP_n.sub file is created for each line in the file, which is used for an integer m iterations, where m is the first item of each line (normally 'X' in CIP)")
parser.add_argument("--cip-explode-jobs",default=None,type=int,help="Number of CIP jobs to use to use in parallel to produce posterior samples for each iteration. Code will generate a fit first, save it, use this number of workers in parallel to generate samples, and then consolidates the samples together. Note that --n-output-samples and --n-eff are NOT adjusted ... if the user wants to adaptively fix the resolution, that needs to be controlled at a higher level")
parser.add_argument("--cip-explode-jobs-last",default=None,type=int,help="Like cip_explode_jobs, but ONLY for last batch. Only applies if using --cip-args-list. Note does NOT adjust control settinggs like n-eff, n-max, so you will need to do this in outside control tools")
parser.add_argument("--cip-explode-jobs-dag",action='store_true',help="If using a worker, that worker has its own node, not a 'queue N' statement on one node. Helps for stability with many workers with high failure probability")
parser.add_argument("--cip-explode-jobs-subdag",action='store_true',help="If using a worker, create a subdag that iterates until the target number of samples are produced, using a POST script on a subdag to repeat effort. Imposes cip_explode_jobs_dag. Note that the actual number used for cip_explode_jobs MAY NOT BE IMPLEMENTED or may be a cap (in progress)")
parser.add_argument("--cip-explode-jobs-flat",action='store_true',help="Pass to use the same arguments for all worker jobs. The main job will be /bin/true.")
parser.add_argument("--puff-exe",default=None,help="util_ParameterPuffball.py")
parser.add_argument("--puff-args",default=None,help="util_ParameterPuffball arguments.  If not specified, puffball will not be performed  ")
parser.add_argument("--puff-cadence",default=None,type=int,help="Every n iterations (not including 0), the puffball code will be applied.  Puffball points will be done *in addition* to the usual results from the DAG.  (The puffball is based on perturbing points from that iteration, and this will roughly double that iteration in ILE job size).  Proposed value 2 (i.e., puff overlap-grid-2, ...-4, ...-6.  If not specified, puffball will not be performed ")
parser.add_argument("--puff-max-it",default=-1,type=int,help="Maximum iteration number that puffball is applied.  If negative, puffball is not applied ")
parser.add_argument("--search-reflected-sky-mode",action='store_true',help="LISA: at one iteration, reflect the CIP-proposed grid's sky to the secondary (reflected) sky mode (the LISA latitude is bimodal), so that iteration explores the reflected mode.")
parser.add_argument("--search-reflected-sky-mode-iteration",default=None,type=int,help="Iteration at which to reflect the sky grid.  Default: n_iterations-2 (the primary mode is mapped by then).")
parser.add_argument("--reflected-sky-mode-exe",default=None,help="convert_primary_sky_mode_to_secondary executable.")
parser.add_argument("--lisa-reference-time",default=0.0,type=float,help="LISA coalescence time (for the reflected-sky transform).")
parser.add_argument("--calmarg-pilot",action='store_true',help="Option C adaptive calibration: add per-iteration cal PILOT jobs (harvest top-lnL points from iteration N's composite, run ILE --calibration-dump-responsibilities, fit+consolidate a cal proposal that SEEDS wide_{N+1} via --calibration-proposal-breadcrumb).  Requires the wide ILE args to already carry the calibration envelope + (per-iteration) proposal-breadcrumb path.")
parser.add_argument("--calmarg-pilot-cadence",default=1,type=int,help="Run a cal pilot every n iterations.  Default 1 (every iteration until the cap).")
parser.add_argument("--calmarg-pilot-max-it",default=3,type=int,help="Stop launching cal pilots after this iteration (cal is boring -> freeze the proposal once learned).  Default 3.")
parser.add_argument("--calmarg-pilot-top-fraction",default=0.05,type=float,help="Fraction of highest-lnL composite points the pilot harvests for full cal.")
parser.add_argument("--calmarg-pilot-max-points",default=32,type=int,help="Cap on harvested pilot points per iteration.")
parser.add_argument("--extrinsic-handoff",action='store_true',help="Extrinsic handoff (GMM sampler): each iteration's wide ILE jobs write a per-event extrinsic proposal (--extrinsic-proposal-output), a per-iteration LOCAL-universe consolidation job (util_ExtrinsicConsolidate.py) picks the most representative one -> extr_consolidated_<it>.npz, which SEEDS wide_{N+1} via --extrinsic-proposal-breadcrumb.  Requires the wide ILE args to already carry those flags (added by util_RIFT_pseudo_pipe.py --extrinsic-handoff).")
parser.add_argument("--extrinsic-handoff-select",default="lnL",help="Metric the extrinsic consolidation ranks per-event proposals by (lnL|neff|n_samples). Default lnL (most peak-representative).")
parser.add_argument("--first-iteration-jumpstart",action='store_true',help="No ILE jobs the first iteration.  Assumes you already have .composite files and want to get going. Particularly helpful for subdag systems")
parser.add_argument("--last-iteration-extrinsic",action='store_true',help="Configure last iteration to extract *one* set of extrinsic parameters from each intrinsic point. [This is highly inefficient, but people like having one extrinsic point per intrinsic point.]  Requires --convert-args")
parser.add_argument("--last-iteration-extrinsic-nsamples",default=3000,type=int,help="Construct this number of extrinsic samples")
parser.add_argument("--last-iteration-extrinsic-samples-per-ile",default=5,type=int,help="Draw this many samples from each ILE job")
parser.add_argument("--last-iteration-extrinsic-samples-per-ile-internal",default=10,type=int,help="Draw this many samples from each ILE job")
parser.add_argument("--last-iteration-extrinsic-batched-convert",action='store_true',help="Used batched converter for output of extrinsic samples")
parser.add_argument("--last-iteration-extrinsic-time-resampling",action='store_true',help="Last iterations use time resampling (+ the fairdraw is done inside ILE itself), so different code path for final stages")
parser.add_argument("--last-iteration-export-marginal-distance-grid", action='store_true', help="Add argument to ILE_extr")
parser.add_argument("--last-iteration-export-distance-slices", default=0, type=int, help="If >0, the ILE_extr (extrinsic) stage exports K-row .dslice files: Plan-B fixed-distance extrinsic-marginalized likelihoods. Adds --export-distance-slices K (+ --internal-use-lnL) to ILE_extr and strips --distance-marginalization.")
parser.add_argument("--last-iteration-export-distance-slices-n-core", default=0, type=int, help="Passthrough to ILE --n-distance-slice-core for the extrinsic-stage .dslice export (0 = ILE default).")
parser.add_argument("--last-iteration-export-distance-slices-n-wing", default=0, type=int, help="Passthrough to ILE --n-distance-slice-wing for the extrinsic-stage .dslice export (0 = ILE default).")
parser.add_argument("--last-iteration-export-distance-slices-all-fresh", action="store_true", default=False, help="Passthrough to ILE --distance-slice-all-fresh: all K slices are fresh fixed-d integrations, no importance-reweight core (honest at low n_eff).")
parser.add_argument("--last-iteration-export-distance-slices-randomize", action="store_true", default=False, help="Passthrough to ILE --distance-slice-randomize: draw each intrinsic's fresh-slice d at random posterior quantiles (K=1 -> fair-draw of d per intrinsic).")
parser.add_argument("--last-iteration-export-distance-slices-wing-neff", default=None, type=int, help="Passthrough to ILE --distance-slice-wing-neff (n_eff per fresh fixed-d slice).")
parser.add_argument("--last-iteration-export-distance-slices-wing-nmax", default=None, type=int, help="Passthrough to ILE --distance-slice-wing-nmax (max samples per fresh fixed-d slice).")
parser.add_argument("--last-iteration-export-distance-slices-wing-delta-lnL", default=None, type=float, help="Passthrough to ILE --distance-slice-wing-delta-lnL for the extrinsic-stage .dslice export.")
parser.add_argument("--last-iteration-export-distance-slices-skip-threshold", default=None, type=float, help="Passthrough to ILE --distance-slice-skip-threshold for the extrinsic-stage .dslice export.")
parser.add_argument("--ile-args",default=None,help="filename of args_ile.txt file  which holds ILE arguments.  Should NOT conflict with arguments auto-set by this DAG ... in particular, i/o arguments will be modified")
parser.add_argument("--ile-exe",default=None,help="filename of ILE or equivalent executable. Will default to `which integrate_likelihood_extrinsic` in low-level code")
parser.add_argument("--ile-retries",default=0,type=int,help="Number of retry attempts for ILE jobs. (These can fail)")
parser.add_argument("--ile-group-subdag",action='store_true',help="If true, ILE jobs are grouped in a subdag. Useful mainly to create iterative DAGs with random ILE jobs/etc")
parser.add_argument("--ile-group-subdag-check-work",action='store_true',help="If true, provides a check for work for each subdag type.")
parser.add_argument("--ile-post-exe",default=None,help="filename of script to execute with SCRIPT POST joblist scripname JOBID RETURN worklevel in the dag. This will be run after each ILE completes in the main dag. Objective is to enable larger-scale CIP queues which work efficiently to a target work level (e.g,, n_eff). Note this script is responsible for reading htcondor logs to identify the run directory!")
parser.add_argument("--ile-condor-commands",default=None,type=str,help="File name for key value pairs for condor commands to add to ILE submit file")
parser.add_argument("--general-retries",default=0,type=int,help="Number of retry attempts for internal jobs (convert, CIP, ...). (These can fail, albeit more rarely, usually due to filesystem problems)")
parser.add_argument("--general-request-disk",default="10M",type=str,help="Request disk passed to condor. Must be done for all jobs now")
parser.add_argument("--ile-request-disk",default="10M",type=str,help="Request disk passed to condor for ILE. Must be done for all jobs now")
parser.add_argument("--cip-request-disk",default="10M",type=str,help="Request disk passed to condor for ILE. Must be done for all jobs now")
parser.add_argument("--cal-request-disk",default="8G",type=str,help="Request disk passed to condor for ILE. Must be done for all jobs now")
parser.add_argument("--ile-n-events-to-analyze",default=1,type=int,help="If >1, you are using ILE_batchmode.  Structures the DAG correctly to account for batch cadence")
parser.add_argument("--ile-n-events-to-analyze-first",default=None,type=int,help="If provided, use a different number batch size for iteration 0. Can be helpful if initial grid is broad.")
parser.add_argument("--ile-runtime-max-minutes",default=None,type=int,help="If not none, kills ILE jobs that take longer than the specified integer number of minutes. Do not use unless an expert")
parser.add_argument("--cip-exe",default=None,help="filename of CIP or equivalent executable. Will default to `which util_ConstructIntrinsicPosterior_GenericCoordinates` in low-level code")
parser.add_argument("--cip-post-exe",default=None,help="filename of script to execute with SCRIPT POST joblist scripname JOBID RETURN worklevel in the dag. This will be run after each CIP completes in the main dag. Objective is to enable larger-scale CIP queues which work efficiently to a target work level (e.g,, n_eff). Note this script is responsible for reading htcondor logs to identify the run directory!")
parser.add_argument("--cip-exe-G",default=None,help="filename of CIP or equivalent executable, as ALTERNATE CIP used when a 'G' iteration is requested ")
parser.add_argument("--use-eccentricity",default=False,action='store_true')
parser.add_argument("--use-meanPerAno",default=False,action='store_true')
parser.add_argument("--use-eccentricity-squared-sampling",default=False,action='store_true')
parser.add_argument("--use-tabular-eos-file",default=False,action='store_true')
parser.add_argument("--test-exe",default=None,help="filename of test code or equivalent executable.  Must have a --test-output <fname> argument.  Used for convergence testing or other termination. NOT ACTIVE; see 'convergence_test_samples.py' for example")
parser.add_argument("--plot-exe",default=None,help="filename of plot code or equivalent executable.  Will default to `which plot_posterior_corner.py`.  Default is to plot last set of samples")
parser.add_argument("--gridinit-exe",default=None,help="filename of initial grid creation or equivalent executable.  Will default to `which util_ManualOverlapGrid.py`.  Must create overlap-grid-0.xml.gz")
parser.add_argument("--frame-rotation",action='store_true',help=" Calculation was done in J frame, not L frame.  Rotate at end. Only if extrinsic samples.")
parser.add_argument("--calibration-reweighting-exe",default=None,help="Calibration reweighting script")
parser.add_argument("--calibration-reweighting",action='store_true',help="Run calibration reweighting on final posterior samples")
parser.add_argument("--calibration-reweighting-batchsize",type=int,default=None,help="If not 'None', tries to group the final set of points based on jobs of a fixed size")
parser.add_argument("--calibration-reweighting-initial-extra-args",type=str,default='',help="Passed to reweighting script as extra args")
parser.add_argument("--calibration-reweighting-extra-args",type=str,default='',help="Passed to the combination script as extra arguments. ")
parser.add_argument("--calibration-reweighting-count",type=int,default=100,help="Number of calibration marginalization realizations. Default is 100")
parser.add_argument("--calibration-reweighting-osg",action='store_true',help="Attempt to use settings for OSG for cal reweighting. Remove after developed")
parser.add_argument("--convert-ascii2h5-exe",default=None,help="Converting ascii to h5 file script")
parser.add_argument("--comov-distance-reweighting-exe",default=None,help="Comoving distance reweighting script")
parser.add_argument("--comov-distance-reweighting",action='store_true',help="Run comoving distance reweighting on final posterior samples")
parser.add_argument("--bilby-pickle-exe",default=None,help="Bilby pickle generation script")
parser.add_argument("--bilby-ini-file",default=None,help="Filename of bilby ini file used to generate pickle file (used for calibration reweighting)")
parser.add_argument("--bilby-pickle-file",default=None,help="Filename of bilby pickle file used for calibration reweighting")
parser.add_argument("--bw-exe",default=None,help="BayesWave location (or wrapper)")
parser.add_argument("--bw-post-exe",default=None,help="BayesWave location (or wrapper)")
parser.add_argument("--fetch-ext-grid-exe",default=None,help="filename of command for fetching external grid. Must have an --output-file argument, which will be auto-set.")
parser.add_argument("--test-args",default=None,help="filename of args_test.txt, which holds test arguments.  Note i/o arguments will be modified, so should NOT specify the samples files or the output file, just the test to be performed and any related arguments")
parser.add_argument("--convert-args",default=None,help="filename of args_convert.txt, which holds arguments to the convert function. Needed if you plan to export tide informaton")
parser.add_argument("--plot-args",default=None,help="filename of args_plot.txt, which holds plot arguments.  Note i/o arguments will be modified, so should NOT specify the samples files or the output file, just the test to be performed and any related arguments.  To create the posterior samples file, you probably need to run the convert job (e.g., via the tests)")
parser.add_argument("--gridinit-args",default=None,help="arguments for grid creation or equivalent executable.  If empty, will attempt to use the --input-grid argument")
parser.add_argument("--fetch-ext-grid-args",default=None,help="arguments for command  for fetching external grid. Don't over-specify the --output-file argument, which will be used by the pipeline")
parser.add_argument("--ile-batch",action='store_true',help="Different workflow: the ILE job is assumed to create a .composite file directly in the current working directory. Designed for fake ILE sets and long-term batch workflow. NOT IMPLEMENTED ")
parser.add_argument("--transfer-file-list",default=None,help="File containing list of *input* filenames to transfer, one name per file. Copied into transfer_files for condor directly.  If provided, also enables attempts to deduce files that need to be transferred for the pipeline to operate, as needed for OSG, etc")
parser.add_argument("--request-memory-ILE",default=4096,type=int,help="Memory request for condor (in Mb) for ILE jobs.")
parser.add_argument("--request-gpu-ILE",action='store_true',help="ILE will request a GPU. [Note: if you do this, your code will only run on GPU-enabled slots]")
parser.add_argument("--ile-gpu-fanout",default=None,help="Multi-GPU ILE fan-out: split each ILE batch's intrinsic-grid range across N GPUs (one shard per GPU). Integer N (also requests N GPUs+CPUs) or 'auto'. Baked into ile_pre.sh; equivalent to RIFT_ILE_GPU_FANOUT. Requires --request-gpu-ILE.")
parser.add_argument("--request-xpu-ILE",action='store_true',help="ILE will *prefer* a gpu, but not require it")
parser.add_argument("--request-memory-CIP",default=16384,type=int,help="Memory request for condor (in Mb) for fitting jobs.")
parser.add_argument("--request-memory-CIP-flex",action='store_true',help="If true,memory request is flexible and will increment. Beware of possible runaway requests")
parser.add_argument("--use-singularity",action='store_true',help="Attempts to use a singularity image in SINGULARITY_RIFT_IMAGE")
parser.add_argument("--use-osg",action='store_true',help="Attempts to set up an OSG workflow.  Must submit from an osg allowed submit machine")
parser.add_argument("--use-osg-cip",action='store_true',help="If use-osg, attempts to run CIP on OSG as well.  Default is to run OSG on local machines")
parser.add_argument("--use-osg-simple-requirements",action='store_true',help="Uses an aggressive simplified requirements string for worker jobs")
parser.add_argument("--condor-local-nonworker",action='store_true',help="Uses local universe for non-worker condor jobs. Important to run in non-NFS location, as other jobs don't have file transfer set up.  Required for public OSG")
parser.add_argument("--condor-containerize-nonworker",action='store_true',help="Provides SingularityImage.  Note the intent is still to use on a *shared* filesystem, not to set up file transfer for the 'nonworker' intermediate glue jobs. Note the default is NOT to use SingularityBindCVMFS right now -- intended for workflows wanting to use a fixed container for all  components")
parser.add_argument("--condor-local-nonworker-igwn-prefix",action='store_true', help="Adds some prefix text to start up cvmfs igwn environment, so local jobs have access to standard RIFT operators. Required for public OSG.")
parser.add_argument("--condor-nogrid-nonworker",action='store_true',help="Uses local flocking for non-worker condor jobs. Important to run in non-NFS location, as other jobs don't have file transfer set up.")
parser.add_argument("--frames-dir",default=None,help="If you are using synthetic dat")
parser.add_argument("--cache-file",default=None,help="If you are using real data and have CVMS frames, you want to transfer the cache file over.")
parser.add_argument("--use-oauth-files",default=None, type=str, help="Authentication method, written into condor file. Valid methods include 'scitokens' " )
parser.add_argument("--use-cvmfs-frames",action='store_true',help="If true, require LIGO frames are present (usually via CVMFS). User is responsible for generating cache file compatible with it.")
parser.add_argument('--n-copies',default=2,type=int,help="Number of duplicates of each ILE job, for redundant Monte Carlo")
parser.add_argument('--n-iterations',default=3,type=int,help="Number of iterations to perform")
parser.add_argument('--n-iterations-subdag-max',default=10,type=int,help="Number of iterations to perform in subdag, maximum ")
parser.add_argument("--start-iteration",default=0,type=int,help="starting iteration. If >0, does not copy over --input-grid. DOES rewrite sub files.  This allows you to change the arguments provided (e.g., use more iterations or settings at late times). Note this overwrites the .sub files ")
parser.add_argument('--n-samples-per-job',default=2000,type=int,help="Number of samples generated each iteration; also, number of ILE jobs run each iteration. Should increase with dimension of problem")
parser.add_argument('--n-samples-per-job-threshold',default=None,type=int,help="If we get this many ILE evaluations, then we have done well enough and can move on; only used with auto-terminating framework.  Note this is applied everywhere")
parser.add_argument('--neff-threshold',default=800,type=int,help="Number of samples generated each iteration")
parser.add_argument("--workflow",default='single',help="[single|fit+posterior|full] describes workflow layout used.  'Single' is a single node, running the fit and posterior for each iteration; 'full' produces many followup jobs to produce a reliable posterior")
parser.add_argument("--n-post-jobs",default=1,type=int,help="Number of posterior jobs. Used in posterior and fit+posterior workflows")
parser.add_argument("--use-bw-psd",action='store_true',help="Use BW PSD, attempting to use fiducial arguments as in LI for placement (i.e., signal at seglen -2). Assumes LI style data convention.  Necessary BW will be parsed out of ile-args.txt (required)")
parser.add_argument("--use-full-submit-paths",action='store_true',help="DAG created has full paths to submit files generated. Note this is implemented on a per-file/as-needed basis, mainly to facilitate using this dag as an external subdag")
opts=  parser.parse_args()

# Multi-GPU ILE fan-out: funnel --ile-gpu-fanout through RIFT_ILE_GPU_FANOUT, which
# dag_utils reads at write_ILE_sub_simple time to size request_GPUs/CPUs and bake the
# value into ile_pre.sh.  CLI value wins over any inherited environment value.
if opts.ile_gpu_fanout is not None:
    os.environ['RIFT_ILE_GPU_FANOUT'] = str(opts.ile_gpu_fanout)


local_worker_universe="vanilla"
local_worker_universe_cip = "vanilla"
no_worker_grid=False
if opts.condor_local_nonworker:
    local_worker_universe="local"
    local_worker_universe_cip="vanilla"  # usually CIP jobs are remote
if opts.condor_nogrid_nonworker:
    no_worker_grid=True

if opts.cip_explode_jobs_subdag:
    opts.cip_explode_jobs_dag =True
    opts.cip_explode_jobs = 2 # nominal
    
extra_text = ''
if opts.condor_local_nonworker_igwn_prefix:
    extra_text += "source /cvmfs/software.igwn.org/conda/etc/profile.d/conda.sh \n"
    extra_text +=  "conda activate igwn \n"

if not opts.cip_explode_jobs_last:  # set realistic default, if not specified
    opts.cip_explode_jobs_last = opts.cip_explode_jobs

working_dir_inside_local = working_dir_inside = opts.working_directory
working_dir_inside_cip = working_dir_inside_local
out_dir_inside_cip = working_dir_inside_local

# ----------------------------------------------------------------------
# Hyperpipeline ASCII format (opt-in via env var).  When set, every
# inter-stage intrinsic-parameter file in the iterative pipeline
# (overlap-grid, puffball, fetch, output-grid) is .dat instead of .xml.gz,
# and ILE consumes them via --sim-grid instead of --sim-xml.  This
# matches the file-naming convention already used by
# create_eos_posterior_pipeline.  When unset, behaviour is identical to
# the legacy pipeline.
#
# We define grid_suffix and sim_grid_flag once here and substitute via
# .format() throughout the rest of the script, so the search-and-replace
# surface area is the literal string "xml.gz" / "--sim-xml" -- nothing
# more invasive than that.
# ----------------------------------------------------------------------
_use_hpip_pipeline = str(os.environ.get("RIFT_HYPERPIPELINE_FORMAT", "")).strip().lower() in ("1", "true", "yes", "on")
grid_suffix = "dat" if _use_hpip_pipeline else "xml.gz"
sim_grid_flag = "--sim-grid" if _use_hpip_pipeline else "--sim-xml"
if _use_hpip_pipeline:
    print(" === Hyperpipeline ASCII grid format active (RIFT_HYPERPIPELINE_FORMAT) ===")
    print("     Inter-stage grids will be .{}, ILE will receive {}".format(grid_suffix, sim_grid_flag))

def _add_hpip_condor_env(job):
    if not _use_hpip_pipeline:
        return
    hpip_env = "RIFT_HYPERPIPELINE_FORMAT=1"
    if hasattr(job, "get_condor_cmds"):
        condor_cmds = job.get_condor_cmds()
    else:
        condor_cmds = getattr(job, "_CondorJob__condor_cmds", {})
    env_cmd = condor_cmds.get("environment")
    if not env_cmd:
        job.add_condor_cmd("environment", '"{}"'.format(hpip_env))
        return
    if hpip_env in env_cmd:
        return
    env_cmd = env_cmd.strip()
    if len(env_cmd) >= 2 and env_cmd[0] == env_cmd[-1] and env_cmd[0] in ("'", '"'):
        env_cmd = env_cmd[:-1] + " " + hpip_env + env_cmd[-1]
    else:
        env_cmd = '"{} {}"'.format(env_cmd, hpip_env)
    job.add_condor_cmd("environment", env_cmd)

if opts.cip_explode_jobs:
    out_dir_inside_cip+= "/iteration_$(macroiteration)_cip/"
os.chdir(opts.working_directory)
# working_dir_inside_local is for jobs on local nodes
if opts.use_singularity or opts.use_osg:
    working_dir_inside = "./" # all files on the remote machine are in the current directory
    if opts.use_osg_cip:
        working_dir_inside_cip = "./"
        out_dir_inside_cip = "./"

singularity_image = None
if opts.use_singularity:
    print(" === USING SINGULARITY === ")
    print(" === container MUST have lal_path2cache ")
    singularity_image = os.environ["SINGULARITY_RIFT_IMAGE"]  # must be present to use singularity
    # SINGULARITY IMAGES ARE ON CVMFS, SO WE CAN AVOID THE SINGULARITY EXEC CALL
    # hardcoding a fiducial copy of lalapps_path2cache; beware about the executable name change
    #os.environ['LALAPPS_PATH2CACHE'] = "/cvmfs/oasis.opensciencegrid.org/ligo/sw/conda/envs/igwn-py39/bin/lal_path2cache" #"singularity exec {singularity_image} lalapps_path2cache".format(singularity_image=singularity_image)
    # REQUIRE CONTAINER HAS THIS ROUTINE ! Watch out
    os.environ['LALAPPS_PATH2CACHE'] = os.environ['SINGULARITY_BASE_EXE_DIR'] + "/lal_path2cache" #"singularity exec {singularity_image} lalapps_path2cache".format(singularity_image=singularity_image)
    print(singularity_image)

    # see https://computing.docs.ligo.org/guide/htcondor/credentials/
    if 'igwn+osdf:' in singularity_image and not opts.use_oauth_files:
        opts.use_oauth_files = 'igwn' # force-set this variable, to save time for end user. This will always be the case
    elif 'osdf:' in singularity_image and not opts.use_oauth_files:
        opts.use_oauth_files = 'scitokens' # force-set this variable, to save time for end user. This will always be the case
    elif not opts.use_oauth_files:
        # Container FAMILY manifest (.yaml/.yml): the per-machine image osdf:// URLs live
        # INSIDE the manifest, so the 'osdf:' substring checks above miss them and the
        # transfer credential is never enabled -> the execute point cannot fetch the
        # selected container ("credential is required for osdf://... but was not
        # discovered" -> all ILE/CIP/CALPILOT jobs held).  Inspect the manifest's image
        # URLs and pick the same credential the single-image path would have.
        try:
            from RIFT.misc.container_manifest import is_container_manifest, load_container_manifest
            if is_container_manifest(singularity_image):
                _imgs = ' '.join(c.get('image', '') for c in load_container_manifest(singularity_image).get('containers', []))
                if 'igwn+osdf:' in _imgs:
                    opts.use_oauth_files = 'igwn'
                elif 'osdf:' in _imgs:
                    opts.use_oauth_files = 'scitokens'
        except Exception:
            pass

if (opts.cip_args is None) and (opts.cip_args_list is None):
    print(" No arguments provided for low-level job")
    sys.exit(0)

# Load args.txt. Remove first item.  Store
if not (opts.cip_args is None):
    with open(opts.cip_args) as f:
        cip_args_list = f.readlines()
    cip_args = ' '.join( [x.replace('\\','') for x in cip_args_list] )
    cip_args = ' '.join(cip_args.split(' ')[1:])
    # Some argument protection for later
    cip_args = cip_args.replace('[', ' \'[')
    cip_args = cip_args.replace(']', ']\'')
    cip_args=cip_args.rstrip()
    cip_args += ' --no-plots '  
    print("CIP", cip_args)

cip_args_lines = None
cip_args_prefixes = []  # so it works correctly even in flat mode
if not (opts.cip_args_list is None):
    with open(opts.cip_args_list) as f:
        cip_args_lines = f.readlines()
    # if one of the prefixes is 'Z', we will be calling a RECURSIVE dag writer, warning!  It will go inside the corresponding CIP directory...
    cip_args_prefixes = [(x.split(' ')[0]) for x in cip_args_lines]  # Pull off the integer
    # 
    cip_args_n = (cip_args_prefixes.copy())
    creating_subdag_list = []
    creating_subdag_arg_list = []
    creating_G_list=[]
    for indx in np.arange(len(cip_args_n)):
        if cip_args_prefixes[indx][0]=='Z':
            creating_subdag_list.append(indx)
            creating_subdag_arg_list.append(' '.join(cip_args_lines[indx].split(' ')[1:]))
            cip_args_n[indx] = 1  # one nominal iteration; the CIP stage in this iteration is replaced by 'run until convergence'
        elif cip_args_prefixes[indx][0] =='G':
            n_G = int(cip_args_prefixes[indx][1:])  # integer after G assumed
            cip_args_n[indx] = int(n_G)
            creating_G_list.append(indx)  # flag for alternate executable
        else:
            cip_args_n[indx]  = int(cip_args_n[indx])
#    cip_args_n = [int(x.split(' ')[0]) for x in cip_args_lines]  # Pull off the integer
    cip_args_lines = [' '.join(x.split(' ')[1:]) for x in cip_args_lines] # pull off the integer
    cip_args_lines = [x.replace('[', ' \'[').replace(']', ']\'').rstrip() for x in cip_args_lines]
    cip_args_lines = [x.lstrip() for x in cip_args_lines] # remove leading whitespace
    for line in cip_args_lines:
        print(" CIP ", line)

    # Above only makes sense if last iteration isn't 'Z'
    if np.sum(cip_args_n) < opts.n_iterations:
        print(" Not enough CIP versions to complete all requested iterations; extending ")
        cip_args_n[-1]  = opts.n_iterations - np.sum(cip_args_n)

# Load args.txt. Remove first item.  Store
with open(opts.ile_args) as f:
    ile_args_list = f.readlines()
ile_args = ' '.join( [x.replace('\\','') for x in ile_args_list] )
ile_args = ' '.join(ile_args.split(' ')[1:])
# Some argument protection for later
ile_args = ile_args.replace('[', ' \'[')
ile_args = ile_args.replace(']', ']\'')
ile_args=ile_args.rstrip()
if opts.request_gpu_ILE or opts.request_xpu_ILE:
    ile_args+=" --vectorized --gpu "  # append this
if opts.ile_n_events_to_analyze > 1:
    ile_args+= " --n-events-to-analyze  $(macrongroup) " # + str(opts.ile_n_events_to_analyze)
#ile_args += ' --soft-fail-event-range '  #important in case event is out of range. Simply exit NO ...not backwards compatible
print("ILE", ile_args)

puff_args=None
puff_cadence = None
puff_max_it = opts.puff_max_it
if opts.puff_args and opts.puff_cadence:
    puff_cadence = opts.puff_cadence
    # Load args.txt. Remove first item.  Store
    with open(opts.puff_args) as f:
        puff_args_list = f.readlines()
    puff_args = ' '.join( [x.replace('\\','') for x in puff_args_list] )
    puff_args = ' '.join(puff_args.split(' ')[1:])
# Some argument protection for later
    puff_args = puff_args.replace('[', ' \'[')
    puff_args = puff_args.replace(']', ']\'')
    puff_args=puff_args.rstrip()
    print("PUFF", puff_args)
    print("PUFF CADENCE", puff_cadence)

# Load args.txt. Remove first item.  Store
fetch_args = None
if opts.fetch_ext_grid_args and opts.fetch_ext_grid_exe:
    with open(opts.fetch_ext_grid_args) as f:
        fetch_args_list = f.readlines()
    fetch_args = ' '.join( [x.replace('\\','') for x in fetch_args_list] )
    fetch_args = ' '.join(fetch_args.split(' ')[1:])
# Some argument protection for later
    fetch_args = fetch_args.replace('[', ' \'[')
    fetch_args = fetch_args.replace(']', ']\'')
    fetch_args=fetch_args.rstrip()
    print("FETCH", fetch_args)


test_args=None
if opts.test_args:
    # Load args.txt. Remove first item.  Store
    with open(opts.test_args) as f:
        test_args_list = f.readlines()
    test_args = ' '.join( [x.replace('\\','') for x in test_args_list] )
    test_args = ' '.join(test_args.split(' ')[1:])
# Some argument protection for later
    test_args = test_args.replace('[', ' \'[')
    test_args = test_args.replace(']', ']\'')
    test_args=test_args.rstrip()
    print("CONVERGE", test_args)

plot_args=None
if opts.plot_args:
    # Load args.txt. Remove first item.  Store
    with open(opts.plot_args) as f:
        plot_args_list = f.readlines()
    plot_args = ' '.join( [x.replace('\\','') for x in plot_args_list] )
    plot_args = ' '.join(plot_args.split(' ')[1:])
# Some argument protection for later
    plot_args = plot_args.replace('[', ' \'[')
    plot_args = plot_args.replace(']', ']\'')
    plot_args=plot_args.rstrip()
    print("PLOT", plot_args)

convert_args=None
if opts.convert_args:
    # Load args.txt. Remove first item.  Store
    with open(opts.convert_args) as f:
        convert_args_list = f.readlines()
    convert_args = ' '.join( [x.replace('\\','') for x in convert_args_list] )
    convert_args = ' '.join(convert_args.split(' ')[1:])
# Some argument protection for later
    convert_args = convert_args.replace('[', ' \'[')
    convert_args = convert_args.replace(']', ']\'')
    convert_args=convert_args.rstrip()
    print("CONVERT", convert_args)

gridinit_args=None
if opts.gridinit_args:
    # Load args.txt. Remove first item.  Store
    with open(opts.gridinit_args) as f:
        gridinit_args_list = f.readlines()
    gridinit_args = ' '.join( [x.replace('\\','') for x in gridinit_args_list] )
    gridinit_args = ' '.join(gridinit_args.split(' ')[1:])
# Some argument protection for later
    gridinit_args = gridinit_args.replace('[', ' \'[')
    gridinit_args = gridinit_args.replace(']', ']\'')
    gridinit_args=gridinit_args.rstrip()
    print("GRIDINIT", gridinit_args)


# Copy seed grid into place as overlap-grid-0.{grid_suffix}
it_start = opts.start_iteration
n_initial = opts.n_samples_per_job
_seed_grid_name = "overlap-grid-0." + grid_suffix
if (it_start == 0) and not gridinit_args:
    shutil.copyfile(opts.input_grid, _seed_grid_name)  # put in working directory !
    if _use_hpip_pipeline:
        # Hyperpipeline files: count data rows = total lines minus comment lines.
        with open(_seed_grid_name) as _f:
            n_initial = sum(1 for _l in _f if _l.strip() and not _l.lstrip().startswith("#"))
        if os.path.lexists("posterior_samples-0.dat"):
            os.remove("posterior_samples-0.dat")
        os.symlink(_seed_grid_name, "posterior_samples-0.dat")
    else:
        n_initial = len(lalsimutils.xml_to_ChooseWaveformParams_array(_seed_grid_name))

transfer_file_names = []
if not (opts.transfer_file_list is None):
    transfer_file_names=[]
    # Load args.txt. Remove first item.  Store
    with open(opts.transfer_file_list) as f:
        for  line in f.readlines():
            transfer_file_names.append(line.rstrip())
    print(" Input files to transfer to job working directory (note!)", transfer_file_names)
# Clean snapshot of the transfer list (PSD + cal envelopes only) BEFORE the grid is
# appended and BEFORE write_ILE_sub_simple mutates it in place (frames_dir, ile_pre.sh).
# The CALPILOT job (built later) needs this clean set, not the wide-ILE-polluted one.
transfer_file_names_pilot_base = list(transfer_file_names)

###
### Fiducial fit job (=sanity check that code will run)
###
if (opts.cip_args is None):
    cip_args = cip_args_lines[0]
cmdname="%s/command-single_fit.sh" % opts.working_directory
cmd = open(cmdname, 'w')
arg_list = cip_args
exe = dag_utils.which("util_ConstructIntrinsicPosterior_GenericCoordinates.py")
cmd.write('#!/usr/bin/env bash\n')
cmd.close()
st = os.stat(cmdname)
import stat
os.chmod(cmdname, st.st_mode | stat.S_IEXEC)


cmdname="%s/command-single.sh" % opts.working_directory
cmd = open(cmdname, 'w')
arg_list = ile_args
arg_list = arg_list.replace('$(macrongroup)','5')
exe = dag_utils.which("integrate_likelihood_extrinsic")
if opts.ile_n_events_to_analyze:
    exe = dag_utils.which("integrate_likelihood_extrinsic_batchmode")
cmd.write('#!/usr/bin/env bash\n')
cmd.write('# (run me only in working directories)\n')
cmd.write(exe + ' ' + arg_list + " {} overlap-grid-0.{}".format(sim_grid_flag, grid_suffix))  # make it concrete, so I can test it
cmd.close()
st = os.stat(cmdname)
import stat
os.chmod(cmdname, st.st_mode | stat.S_IEXEC)
# Add argument to ile_args.txt to 
#    identify overlap grid input
#    identify output file names (?)
ile_args_orig = ile_args  # provides ability to strip out the output and replace it with alternate
ile_args+= ' {} '.format(sim_grid_flag) + working_dir_inside + '/overlap-grid-$(macroiteration).{} '.format(grid_suffix)
ile_args_forpuff= ile_args_orig + ' {} '.format(sim_grid_flag) + working_dir_inside + '/puffball-$(macroiteration).{} '.format(grid_suffix)
ile_args_forfetch= ile_args_orig + ' {} '.format(sim_grid_flag) + working_dir_inside + '/fetch-$(macroiteration).{} '.format(grid_suffix)


###
### DAG generation
###

dag = pipeline.CondorDAG(log=os.getcwd())


# Make PSD, if requested
if opts.use_bw_psd:
    print(" ===> Adding BW PSD to pipeline <=== ")
    opts_ile = parse_ile_args_for_bw(ile_args_orig)

    channel_names = {}
    for inst, chan in [c.split("=") for c in opts_ile.channel_name]:
        channel_names[inst] = chan

    flow_ifo_dict = {}
    if opts_ile.fmin_ifo:
        for inst, freq_str in [c.split("=") for c in opts_ile.fmin_ifo]:
           flow_ifo_dict[inst]=float(freq_str) 

    
    channel_dict_bw = {}
    for ifo in list(channel_names.keys()):
        fmin_here= opts_ile.fmin_template
        if ifo in list(flow_ifo_dict.keys()):
            fmin_here  = flow_ifo_dict[ifo]
        channel_dict_bw[ifo]= [channel_names[ifo], fmin_here]
    
    # make BW directory
    bw_dir = opts.working_directory+"/make_bw_psds"
    mkdir(bw_dir)
    for ifo in list(channel_names.keys()):
        mkdir(bw_dir+"/"+ifo)

    # seglen computation
    seglen  = opts_ile.data_end_time - opts_ile.data_start_time

    # Write bw sub files
    bw_universe='local'
    if no_worker_grid:
        bw_universe='vanilla'
    bw_job, bw_job_name = dag_utils.write_psd_sub_BW_monoblock(cache_file=opts_ile.cache_file,psd_length=seglen,srate=opts_ile.srate,event_time=opts_ile.event_time,log_dir=bw_dir+"/",exe=opts.bw_exe,no_grid=no_worker_grid,universe=bw_universe)
    bw_job.add_condor_cmd("initialdir",bw_dir+"/$(ifo)")
    bw_job.add_condor_cmd('request_disk',opts.general_request_disk)
    if opts.use_full_submit_paths:
        fname = opts.working_directory+"/"+bw_job.get_sub_file()
        bw_job.set_sub_file(fname)
    bw_job.write_sub_file()

    # bw_job, bw_job_name = dag_utils.write_psd_sub_BW_step0(channel_dict=channel_dict_bw,cache_file=opts_ile.cache_file,psd_length=seglen,srate=opts_ile.srate,event_time=opts_ile.event_time,log_dir=bw_dir+"/",exe=opts.bw_exe)
    # bw2_job, bw2_job_name=dag_utils.write_psd_sub_BW_step1(channel_dict=channel_dict_bw,cache_file=opts_ile.cache_file,psd_length=seglen,srate=opts_ile.srate,event_time=opts_ile.event_time,log_dir=bw_dir+"/",exe=opts.bw_post_exe)
    # bw_job.add_condor_cmd("initialdir",bw_dir)
    # bw2_job.add_condor_cmd("initialdir",bw_dir)
    # bw2_job.write_sub_file()

    # write convert command
#    event_time_trunc ='%.2f' % float(opts_ile.event_time)   # format so only 3 digits after decimal 
#    event_time_trunc += '0'  # add last character, always zero?
#    convert_psd_job, convert_psd_job_name = dag_utils.write_convertpsd_sub(ifo="$(ifo)",file_input=bw_dir+"/" + str(event_time_trunc) + "_$(ifo)-PSD.dat",log_dir=bw_dir+"/")
    convert_psd_job, convert_psd_job_name = dag_utils.write_convertpsd_sub(ifo="$(ifo)",file_input=bw_dir+"/$(ifo)/$(ifo)_fairdraw_psd.dat",log_dir=bw_dir+"/$(ifo)/")
    convert_psd_job.add_condor_cmd('initialdir',working_dir_inside)
    convert_psd_job.add_condor_cmd('request_disk',opts.general_request_disk)
    if opts.use_full_submit_paths:
        fname = opts.working_directory+"/"+convert_psd_job.get_sub_file()
        convert_psd_job.set_sub_file(fname)
    convert_psd_job.write_sub_file()


# Make directories for all iterations
subdag_dir = opts.working_directory+"/various_subdags"
mkdir(subdag_dir)
for indx in np.arange(it_start,opts.n_iterations+1):
    ile_dir = opts.working_directory+"/iteration_"+str(indx)+"_ile"
    cip_dir = opts.working_directory+"/iteration_"+str(indx)+"_cip"
    consolidate_dir = opts.working_directory+"/iteration_"+str(indx)+"_con"
#    convert_dir = opts.working_directory+"/iteration_"+str(indx)+"_change"
    mkdir(ile_dir); mkdir(ile_dir+"/logs")
    mkdir(cip_dir);  mkdir(cip_dir+"/logs")
    mkdir(consolidate_dir); mkdir(consolidate_dir+"/logs")
#    mkdir(change_dir); mkdir(change_dir+"/logs")

    if opts.test_args:
        test_dir = opts.working_directory+"/iteration_"+str(indx)+"_test"
        mkdir(test_dir); mkdir(test_dir+'/logs')
        
    if opts.plot_args:  # Overkill: currently only making plots on last iteration
        plot_dir = opts.working_directory+"/iteration_"+str(indx)+"_plot"
        mkdir(plot_dir); mkdir(plot_dir+'/logs')

# make calmarg directory and log directory
if opts.calibration_reweighting:
    cal_dir = opts.working_directory+"/calmarg"
    mkdir(cal_dir)
    mkdir(cal_dir+"/logs")

# ++++
# Create workflow tasks
# ++++

##   ILE job

ile_exe =opts.ile_exe
ile_condor_commands=None
if opts.ile_condor_commands and os.path.exists(opts.ile_condor_commands):
    lines =None
    ile_condor_commands = {}
    with open(opts.ile_condor_commands, 'r') as f:
      for line in f:
        key, value = line.split(" ", 1) # force a single splitf
        ile_condor_commands[key] = value
    # reset to None if none provided
    if len(ile_condor_commands) < 1:
        ile_condor_commands = None
if (opts.ile_n_events_to_analyze > 1) and (exe is None):
    ile_exe = dag_utils.which("integrate_likelihood_extrinsic_batchmode")
output_file_names = None
request_disk=opts.ile_request_disk
if opts.use_singularity or opts.use_osg:
    transfer_file_names.append("../overlap-grid-$(macroiteration).{}".format(grid_suffix))
    #request_disk=4
    #output_file_names = ','.join(["CME_out-$(macroevent)-$(cluster)-$(process).xml_{0}_.dat".format(x) for x in np.arange(opts.ile_n_events_to_analyze)])
    #print "OUTPUT FILES ", output_file_names
ile_job, ile_job_name = dag_utils.write_ILE_sub_simple(tag='ILE',log_dir=None,arg_str=ile_args,output_file="CME_out.xml",ncopies=opts.n_copies,exe=ile_exe,transfer_files=transfer_file_names,transfer_output_files=output_file_names,request_memory=opts.request_memory_ILE,request_disk=request_disk,request_gpu=opts.request_gpu_ILE,request_cross_platform=opts.request_xpu_ILE,use_singularity=opts.use_singularity,singularity_image=singularity_image,use_osg=opts.use_osg,simple_osg_requirements=opts.use_osg_simple_requirements,frames_dir=opts.frames_dir,cache_file=opts.cache_file,use_cvmfs_frames=opts.use_cvmfs_frames,use_oauth_files=opts.use_oauth_files,max_runtime_minutes=opts.ile_runtime_max_minutes,condor_commands=ile_condor_commands)
# Modify: create macro for iteration 
#   - added on a per-node basis
# Modify: add macro argument for overlap grid to be used (kept in top-level directory)
# Modify: set 'initialdir'
#     NOTE: logs will be specified relative to this directory
ile_job.add_condor_cmd("initialdir",opts.working_directory+"/iteration_$(macroiteration)_ile")
# Modify output argument: change logs and working directory to be subdirectory for the run
ile_job.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/ILE-$(macroevent)-$(cluster)-$(process).log")
ile_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/ILE-$(macroevent)-$(cluster)-$(process).err")
ile_job.set_stdout_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/ILE-$(macroevent)-$(cluster)-$(process).out")
if opts.use_full_submit_paths:
    fname = opts.working_directory+"/"+ile_job.get_sub_file()
    ile_job.set_sub_file(fname)
_add_hpip_condor_env(ile_job)
ile_job.write_sub_file()

if not (opts.puff_args is None):
    transfer_file_names_puff = list(transfer_file_names[:-1])
    transfer_file_names_puff.append(opts.working_directory+"/puffball-$(macroiteration).{}".format(grid_suffix)) # could also use ../puffball, because relative is ok to initialdir
    # Write a version of the ILE job that uses puffball inputs ... IDENTICAL to code above for standard ILE case,just different argument for input
    # Yes, it would logically be simpler to make overlap-grid.xml.gz larger ... but I want to keep 'real samples' and 'puffed samples' seperate.
    ilePuff_job, ilePuff_job_name = dag_utils.write_ILE_sub_simple(tag='ILE_puff',log_dir=None,arg_str=ile_args_forpuff,output_file="CME_puff_out.xml",ncopies=opts.n_copies,exe=ile_exe,transfer_files=transfer_file_names_puff,request_memory=opts.request_memory_ILE,request_disk=request_disk,request_gpu=opts.request_gpu_ILE,request_cross_platform=opts.request_xpu_ILE,use_singularity=opts.use_singularity,singularity_image=singularity_image,use_osg=opts.use_osg,simple_osg_requirements=opts.use_osg_simple_requirements,frames_dir=opts.frames_dir,cache_file=opts.cache_file,use_cvmfs_frames=opts.use_cvmfs_frames,use_oauth_files=opts.use_oauth_files,max_runtime_minutes=opts.ile_runtime_max_minutes,condor_commands=ile_condor_commands)
    ilePuff_job.add_condor_cmd("initialdir",opts.working_directory+"/iteration_$(macroiteration)_ile")
    ilePuff_job.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/ILE_puff-$(macroevent)-$(cluster)-$(process).log")
    ilePuff_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/ILE_puff-$(macroevent)-$(cluster)-$(process).err")
    ilePuff_job.set_stdout_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/ILE_puff-$(macroevent)-$(cluster)-$(process).out")
    if opts.use_full_submit_paths:
        fname = opts.working_directory+"/"+ilePuff_job.get_sub_file()
        ilePuff_job.set_sub_file(fname)
    _add_hpip_condor_env(ilePuff_job)
    ilePuff_job.write_sub_file()

if not (fetch_args is None):
    transfer_file_names_fetch = list(transfer_file_names[:-1])
    transfer_file_names_fetch.append(opts.working_directory+"/fetch-$(macroiteration).{}".format(grid_suffix)) # could also use ../puffball, because relative is ok to initialdir
    # Write a version of the ILE job that uses puffball inputs ... IDENTICAL to code above for standard ILE case,just different argument for input
    # Yes, it would logically be simpler to make overlap-grid.xml.gz larger ... but I want to keep 'real samples' and 'puffed samples' seperate.
    ileFetch_job, ileFetch_job_name = dag_utils.write_ILE_sub_simple(tag='ILE_fetch',log_dir=None,arg_str=ile_args_forfetch,output_file="CME_fetch_out.xml",ncopies=opts.n_copies,exe=ile_exe,transfer_files=transfer_file_names_fetch,request_memory=opts.request_memory_ILE,request_disk=request_disk,request_gpu=opts.request_gpu_ILE,request_cross_platform=opts.request_xpu_ILE,use_singularity=opts.use_singularity,singularity_image=singularity_image,use_osg=opts.use_osg,simple_osg_requirements=opts.use_osg_simple_requirements,frames_dir=opts.frames_dir,cache_file=opts.cache_file,use_cvmfs_frames=opts.use_cvmfs_frames,oauth_files=opts.use_oauth_files,max_runtime_minutes=opts.ile_runtime_max_minutes,condor_commands=ile_condor_commands)
    ileFetch_job.add_condor_cmd("initialdir",opts.working_directory+"/iteration_$(macroiteration)_ile")
    ileFetch_job.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/ILE_fetch-$(macroevent)-$(cluster)-$(process).log")
    ileFetch_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/ILE_fetch-$(macroevent)-$(cluster)-$(process).err")
    ileFetch_job.set_stdout_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/ILE_fetch-$(macroevent)-$(cluster)-$(process).out")
    if opts.use_full_submit_paths:
        fname = opts.working_directory+"/"+ileFetch_job.get_sub_file()
        ileFetch_job.set_sub_file(fname)
    _add_hpip_condor_env(ileFetch_job)
    ileFetch_job.write_sub_file()


if  (opts.last_iteration_extrinsic):
    n_points_per_ILE = opts.last_iteration_extrinsic_samples_per_ile
    neff_orig = parse_ile_args_lazy(ile_args_orig,'n-eff') # parse to get originally requested n_eff, as string
    print(" Orig n_eff ", neff_orig, n_points_per_ILE)
    n_eff_last = np.min([2*n_points_per_ILE,opts.last_iteration_extrinsic_samples_per_ile])  # make n_eff large enough to get all samples requested. Provide cap in case user requests very vew extrinsic workers.
    if isinstance(neff_orig,str):
        n_eff_last  = np.max([n_eff_last, int(neff_orig)]) 
    # # ILE job with modified output format
    # #  - note we *double* the memory request, because we need space to save samples
    # #  - note
    # opts_ile = parse_ile_args_flexible(ile_args_orig) # parse to get originally requested n_eff, cache-file
    # n_eff_last = np.min([2*n_points_per_ILE,opts.last_iteration_extrinsic_samples_per_ile])  # make n_eff large enough to get all samples requested. Provide cap in case user requests very vew extrinsic workers.
    # if hasattr(opts_ile, 'n_eff'):
    #     n_eff_last  = np.max([n_eff_last, opts_ile.n_eff]) # NEVER USE FEWER n_eff THAN WAS USED FOR intrinsic run. Fallback to no change in case (never happens) not in arg list
    ile_args_extr = ile_args + " --save-P 0.01 --save-samples --n-eff  "  +str(n_eff_last)  # modify convergence criteria so output of reasonable size/matching needs of extrinsic output
    #  - note we *disable* --no-adapt-after-first (if present), so each point is independent (e.g., in sky location)
    ile_args_extr = ile_args_extr.replace('--no-adapt-after-first','')
    if opts.last_iteration_export_marginal_distance_grid:
        ile_args_extr += " --export-marginal-distance-grid "
        ile_args_extr = ile_args_extr.replace("--distance-marginalization ", ' ')  # *currently* cannot use distance marginalization in the last step if we want distance grid output
    if opts.last_iteration_export_distance_slices and opts.last_iteration_export_distance_slices > 0:
        # Plan-B distance slices: K independent fixed-d extrinsic integrals per
        # intrinsic point, emitted as a .dslice file. Requires lnL mode and no
        # distance marginalization (same constraints as the grid export above).
        ile_args_extr += " --export-distance-slices {} ".format(opts.last_iteration_export_distance_slices)
        if opts.last_iteration_export_distance_slices_all_fresh:
            ile_args_extr += " --distance-slice-all-fresh "
        if opts.last_iteration_export_distance_slices_randomize:
            ile_args_extr += " --distance-slice-randomize "
        if opts.last_iteration_export_distance_slices_wing_neff is not None:
            ile_args_extr += " --distance-slice-wing-neff {} ".format(opts.last_iteration_export_distance_slices_wing_neff)
        if opts.last_iteration_export_distance_slices_wing_nmax is not None:
            ile_args_extr += " --distance-slice-wing-nmax {} ".format(opts.last_iteration_export_distance_slices_wing_nmax)
        if opts.last_iteration_export_distance_slices_n_core:
            ile_args_extr += " --n-distance-slice-core {} ".format(opts.last_iteration_export_distance_slices_n_core)
        if opts.last_iteration_export_distance_slices_n_wing:
            ile_args_extr += " --n-distance-slice-wing {} ".format(opts.last_iteration_export_distance_slices_n_wing)
        if opts.last_iteration_export_distance_slices_wing_delta_lnL is not None:
            ile_args_extr += " --distance-slice-wing-delta-lnL {} ".format(opts.last_iteration_export_distance_slices_wing_delta_lnL)
        if opts.last_iteration_export_distance_slices_skip_threshold is not None:
            ile_args_extr += " --distance-slice-skip-threshold {} ".format(opts.last_iteration_export_distance_slices_skip_threshold)
        if "--internal-use-lnL" not in ile_args_extr:
            ile_args_extr += " --internal-use-lnL "
        ile_args_extr = ile_args_extr.replace("--distance-marginalization ", ' ')
    if opts.last_iteration_extrinsic_time_resampling:
        ile_args_extr += " --resample-time-marginalization --fairdraw-extrinsic-output --fairdraw-extrinsic-output-n-max {} ".format(n_points_per_ILE)
        print(" Time resampling in extraction iteration: **DISABLING** distance marginalization if present ")
        ile_args_extr = ile_args_extr.replace("--distance-marginalization ", ' ')  # *currently* cannot use distance marginalization in the last step if we want fairdraw output.  Note the *unfairdraw* output can ignore this ...
    ileExtr_job, ileExtr_job_name = dag_utils.write_ILE_sub_simple(tag='ILE_extr',log_dir=None,arg_str=ile_args_extr,output_file="EXTR_out.xml",simple_unique=True,ncopies=1,exe=ile_exe,transfer_files=transfer_file_names,request_memory=opts.request_memory_ILE*2,request_gpu=opts.request_gpu_ILE,request_cross_platform=opts.request_xpu_ILE,use_cvmfs_frames=opts.use_cvmfs_frames,request_disk=request_disk,use_singularity=opts.use_singularity,singularity_image=singularity_image,use_osg=opts.use_osg,simple_osg_requirements=opts.use_osg_simple_requirements,use_oauth_files=opts.use_oauth_files,frames_dir=opts.frames_dir,cache_file=opts.cache_file)
    ileExtr_job.add_condor_cmd("initialdir",opts.working_directory+"/iteration_$(macroiteration)_ile")
    ileExtr_job.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/ILEextr-$(macroevent)-$(cluster)-$(process).log")
    ileExtr_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/ILEextr-$(macroevent)-$(cluster)-$(process).err")
    ileExtr_job.set_stdout_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/ILEextr-$(macroevent)-$(cluster)-$(process).out")
    if opts.use_full_submit_paths:
        fname = opts.working_directory+"/"+ileExtr_job.get_sub_file()
        ileExtr_job.set_sub_file(fname)
    ileExtr_job.write_sub_file()
    
    # Convert task 
    if opts.last_iteration_extrinsic_time_resampling:
        # very simple: igwn_ligolw add on all final output, then a single convert
        convert_args_extr =  " --convention LI  --export-cosmology --use-interpolated-cosmology " 
        if not (convert_args is None):
            convert_args_extr += convert_args
        # batched convert shell script
        relevant_path = dag_utils.which('util_JoinExtrXML.py')
        relevant_path_2 = dag_utils.which('convert_output_format_ile2inference')
        if opts.condor_local_nonworker_igwn_prefix:
            # we are in igwn environment
            relevant_path = 'util_JoinExtrXML.py'  
            relevant_path_2 = 'convert_output_format_ile2inference'
        # randomization: 'shuf' is preferred, but otherwise use ' sort -R'. Note performed locally, so local filesystem/os is fine.
        extra_shuffle_command = ' | cat'
        which_shuf = which('shuf'); which_sort = which('sort')
        if isinstance(which_shuf, str):
            extra_shuffle_command = ' | {} '.format(which_shuf)
        elif isinstance(which_sort, str):
            extra_shuffle_command  = ' | {} -R '.format(which_sort)
        with open("allinone_convert.sh",'w') as f:
            f.write(f"""#! /bin/bash
{extra_text}
{relevant_path} ./iteration_$1_ile/'EXTR_out-*.xml_*_.xml.gz' --output ./tmp_converted.xml.gz
{relevant_path_2} {convert_args_extr} ./tmp_converted.xml.gz  > ./tmp_converted.dat
head -n 1 ./tmp_converted.dat > ./extrinsic_posterior_samples.dat
sed 1d ./tmp_converted.dat  {extra_shuffle_command} >> ./extrinsic_posterior_samples.dat
""")#.format(extra_text,relevant_path,'.','.',relevant_path_2,convert_args_extr,opts.working_directory,extra_shuffle_command,opts.working_directory))
        os.system("chmod a+x allinone_convert.sh")
        # Create job submit file
        batchConvertExtr_job, batchConvertExtr_job_name = dag_utils.write_convert_sub(exe=opts.working_directory+"/allinone_convert.sh",tag='convert_extr',log_dir=None,arg_str='',file_input="$(macroiteration) ",file_output="/dev/null", out_dir=opts.working_directory,universe=local_worker_universe,no_grid=no_worker_grid)
        batchConvertExtr_job.add_condor_cmd("initialdir",opts.working_directory)
        batchConvertExtr_job.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/batchconvert-$(macroevent).log")
        batchConvertExtr_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/batchconvert-$(macroevent).err")
        batchConvertExtr_job.add_condor_cmd('request_disk',opts.general_request_disk)
        if opts.use_full_submit_paths:
            fname = opts.working_directory+"/"+batchConvertExtr_job.get_sub_file()
            batchConvertExtr_job.set_sub_file(fname)
        batchConvertExtr_job.write_sub_file()
    elif opts.last_iteration_extrinsic_batched_convert:
        # Batched conversion: one job for a great many items
        convert_args_extr =  " --convention LI  --export-cosmology --use-interpolated-cosmology " 
        if opts.last_iteration_extrinsic_samples_per_ile:
            convert_args_extr += " --n-output-samples-per-file {}".format(opts.last_iteration_extrinsic_samples_per_ile)
        if not (convert_args is None):
            convert_args_extr += convert_args
        relevant_path = dag_utils.which('util_BatchConvertResampleILEOutput.py')
        if opts.condor_local_nonworker_igwn_prefix:
            # we are in igwn environment
            relevant_path = 'util_BatchConvertResampleILEOutput.py'
        # batched convert shell script
        with open("batch_convert.sh",'w') as f:
            f.write("""#! /bin/bash
{}
{} {}/iteration_$1_ile/EXTR_out-$2.xml_*_.xml.gz {}
""".format(extra_text,relevant_path,opts.working_directory,convert_args_extr))
        os.system("chmod a+x batch_convert.sh")
        # Create job submit file
        batchConvertExtr_job, batchConvertExtr_job_name = dag_utils.write_convert_sub(exe=opts.working_directory+"/batch_convert.sh",tag='convert_extr',log_dir=None,arg_str='',file_input="$(macroiteration) $(macroevent)",file_output=opts.working_directory+"/iteration_$(macroiteration)_ile/EXTR_out-$(macroevent).downsampled_dat.dat", out_dir=opts.working_directory+"/iteration_$(macroiteration)_ile/",universe=local_worker_universe,no_grid=no_worker_grid)
        batchConvertExtr_job.add_condor_cmd("initialdir",opts.working_directory)
        batchConvertExtr_job.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/batchconvert-$(macroevent).log")
        batchConvertExtr_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/batchconvert-$(macroevent).err")
        batchConvertExtr_job.add_condor_cmd('request_disk',opts.general_request_disk)
        if opts.use_full_submit_paths:
            fname = opts.working_directory+"/"+batchConvertExtr_job.get_sub_file()
            batchConvertExtr_job.set_sub_file(fname)
        batchConvertExtr_job.write_sub_file()
    else:
        # Old style convert
        convert_args_extr =  " --convention LI  --export-cosmology --use-interpolated-cosmology " 
        if not (convert_args is None):
            convert_args_extr += convert_args
        convertExtr_job, convertExtr_job_name = dag_utils.write_convert_sub(tag='convert_extr',log_dir=None,arg_str=convert_args_extr,file_input=opts.working_directory+"/iteration_$(macroiteration)_ile/EXTR_out-$(macroevent).xml_$(macroindx)_.xml.gz",file_output=opts.working_directory+"/iteration_$(macroiteration)_ile/EXTR_out-$(macroevent).xml_$(macroindx)_.dat", out_dir=opts.working_directory+"/iteration_$(macroiteration)_ile/",universe=local_worker_universe,no_grid=no_worker_grid)
        convertExtr_job.add_condor_cmd("initialdir",opts.working_directory)
        convertExtr_job.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/convert-$(macroevent)-$(macroindx).log")
        convertExtr_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/convert-$(macroevent)-$(macroindx).err")
        convertExtr_job.add_condor_cmd('request_disk',opts.general_request_disk)
        if opts.use_full_submit_paths:
            fname = opts.working_directory+"/"+convertExtr_job.get_sub_file()
            convertExtr_job.set_sub_file(fname)
        convertExtr_job.write_sub_file()

        # Resample task 
        resample_args = ' --n-output-samples  ' + str(n_points_per_ILE)  # pick 5 random points from each ILE run
        resample_job, resample_job_name = dag_utils.write_resample_sub('resample',log_dir=None,arg_str=resample_args,file_input=opts.working_directory+"/iteration_$(macroiteration)_ile/EXTR_out-$(macroevent).xml_$(macroindx)_.dat",file_output=opts.working_directory+"/iteration_$(macroiteration)_ile/EXTR_out-$(macroevent).xml_$(macroindx)_.downsampled_dat",universe=local_worker_universe,no_grid=no_worker_grid)
        resample_job.add_condor_cmd("initialdir",opts.working_directory)
        resample_job.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/resample-$(macroevent)-$(macroindx).log")
        resample_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/resample-$(macroevent)-$(macroindx).err")
        resample_job.add_condor_cmd('request_disk',opts.general_request_disk)
        if opts.use_full_submit_paths:
            fname = opts.working_directory+"/"+resample_job.get_sub_file()
            resample_job.set_sub_file(fname)
        resample_job.write_sub_file()

    # Combination task at end -- probably should be a general utility
    cat_job, cat_job_name = dag_utils.write_cat_sub(file_prefix='EXTR', file_postfix='.downsampled_dat.dat',file_output='extrinsic_posterior_samples.dat',universe=local_worker_universe,no_grid=no_worker_grid)
    cat_job.add_condor_cmd("initialdir",opts.working_directory)
    cat_job.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/cat-$(cluster)-$(process).log")
    cat_job.set_stdout_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/cat-$(cluster)-$(process).out")
    cat_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_ile/logs/cat-$(cluster)-$(process).err")
    cat_job.add_condor_cmd('request_disk',opts.general_request_disk)
    if opts.use_full_submit_paths:
        fname = opts.working_directory+"/"+cat_job.get_sub_file()
        cat_job.set_sub_file(fname)
    cat_job.write_sub_file()

    # Per-distance likelihood export: consolidate per-event .dgrid (Plan A)
    # and/or .dslice (Plan B) files into a single table at the run root.  The
    # consolidated file is the "net" intrinsic+distance grid downstream tools
    # like util_ConstructEOSPosterior.py consume.  Each consolidation is
    # gated on the corresponding --last-iteration-export-* flag so we never
    # emit an empty sub for runs that did not request the export.
    if opts.last_iteration_export_marginal_distance_grid:
        dgrid_job, dgrid_job_name = dag_utils.write_consolidate_distance_grids_sub(
            tag='consolidate_dgrid',
            # Match the real per-event batched-extrinsic outputs
            # EXTR_out-<event>.xml_<k>_.dgrid (note the leading-segment wildcard;
            # 'EXTR_out.xml_*_.dgrid' has no '-<event>' infix and matches nothing).
            input_glob='EXTR_out*.xml_*_.dgrid',
            file_output=opts.working_directory + '/all_dgrid.dat',
            # iteration_*_ile wildcard, NOT $(macroiteration): the consolidate driver
            # is a shell script (no condor-macro substitution); the wildcard matches
            # the final-iteration directory regardless of its number.
            search_dir=opts.working_directory + '/iteration_*_ile',
            log_dir=opts.working_directory + '/iteration_$(macroiteration)_ile/logs/',
            universe=local_worker_universe, no_grid=no_worker_grid,
        )
        dgrid_job.add_condor_cmd("initialdir", opts.working_directory)
        dgrid_job.add_condor_cmd('request_disk', opts.general_request_disk)
        if opts.use_full_submit_paths:
            fname = opts.working_directory + "/" + dgrid_job.get_sub_file()
            dgrid_job.set_sub_file(fname)
        dgrid_job.write_sub_file()
    if opts.last_iteration_export_distance_slices and opts.last_iteration_export_distance_slices > 0:
        dslice_job, dslice_job_name = dag_utils.write_consolidate_distance_grids_sub(
            tag='consolidate_dslice',
            input_glob='EXTR_out*.xml_*_.dslice',
            file_output=opts.working_directory + '/all_dslice.dat',
            search_dir=opts.working_directory + '/iteration_*_ile',
            log_dir=opts.working_directory + '/iteration_$(macroiteration)_ile/logs/',
            universe=local_worker_universe, no_grid=no_worker_grid,
        )
        dslice_job.add_condor_cmd("initialdir", opts.working_directory)
        dslice_job.add_condor_cmd('request_disk', opts.general_request_disk)
        if opts.use_full_submit_paths:
            fname = opts.working_directory + "/" + dslice_job.get_sub_file()
            dslice_job.set_sub_file(fname)
        dslice_job.write_sub_file()


##   Consolidate job(s)
#   - consolidate output of single previous job
con_arg_str = ''
if opts.use_eccentricity:
    con_arg_str += " --eccentricity "
    if opts.use_meanPerAno:
        con_arg_str += " --meanPerAno "
if opts.use_tabular_eos_file:
    con_arg_str += " --tabular-eos-file "
con_job, con_job_name = dag_utils.write_consolidate_sub_simple(tag='join',log_dir=None,arg_str=con_arg_str,base=opts.working_directory+"/iteration_$(macroiteration)_ile", target=opts.working_directory+'/consolidated_$(macroiteration)',universe=local_worker_universe,no_grid=no_worker_grid,extra_text=extra_text)
# Modify: set 'initialdir' : should run in top-level direcory
#con_job.add_condor_cmd("initialdir",opts.working_directory+"/iteration_$(macroiteration)_con")
# Modify output argument: change logs and working directory to be subdirectory for the run
con_job.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_con/logs/con-$(cluster)-$(process).log")
con_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_con/logs/con-$(cluster)-$(process).err")
con_job.set_stdout_file(opts.working_directory+"/iteration_$(macroiteration)_con/logs/con-$(cluster)-$(process).out")
con_job.add_condor_cmd('request_disk',opts.general_request_disk)
if opts.use_full_submit_paths:
    fname = opts.working_directory+"/"+con_job.get_sub_file()
    con_job.set_sub_file(fname)
if opts.condor_containerize_nonworker and singularity_image:
    con_job.add_condor_cmd("+SingularityImage", '"' + singularity_image + '"')
_add_hpip_condor_env(con_job)
con_job.write_sub_file()

##   Unify job
#   - update 'all.net' to include all previous events
unify_arg_str = ''
if opts.use_eccentricity:
    unify_arg_str += " --eccentricity "
    if opts.use_meanPerAno:
        unify_arg_str += " --meanPerAno "
if opts.use_tabular_eos_file:
    unify_arg_str += " --tabular-eos-file "
unify_job, unify_job_name = dag_utils.write_unify_sub_simple(tag='unify',log_dir='',arg_str=unify_arg_str, base=opts.working_directory, target=opts.working_directory+'/all.net',universe=local_worker_universe,no_grid=no_worker_grid,extra_text=extra_text)
unify_job.add_condor_cmd("initialdir",opts.working_directory)
unify_job.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_con/logs/unify-$(cluster)-$(process).log")
unify_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_con/logs/unify-$(cluster)-$(process).err")
unify_job.set_stdout_file(opts.working_directory+"/all.net")
unify_job.add_condor_cmd('request_disk',opts.general_request_disk)
if opts.use_full_submit_paths:
    fname = opts.working_directory+"/"+unify_job.get_sub_file()
    unify_job.set_sub_file(fname)
if opts.condor_containerize_nonworker and singularity_image:
    unify_job.add_condor_cmd("+SingularityImage", '"' + singularity_image + '"')
_add_hpip_condor_env(unify_job)
unify_job.write_sub_file()


# Evidence job
exe_evidence = which('util_CIPDirSummarizeEvidence.py')
evidence_job, evidence_job_name = dag_utils.write_convert_sub(tag='evidence', exe=exe_evidence, log_dir='', file_input='',  arg_str=" --cip-dir iteration_$(macroiteration)_cip --output evidence_$(macroiteration) ",  universe=local_worker_universe, no_grid=no_worker_grid)
evidence_job.add_condor_cmd("initialdir",opts.working_directory)
evidence_job.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_con/logs/evidence-$(cluster)-$(process).log")
evidence_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_con/logs/evidence-$(cluster)-$(process).err")
evidence_job.set_stdout_file(opts.working_directory+"/iteration_$(macroiteration)_con/logs/evidence-$(cluster)-$(process).out")
evidence_job.add_condor_cmd('request_disk',opts.general_request_disk)
if opts.use_full_submit_paths:
    fname = opts.working_directory+"/"+evidence_job.get_sub_file()
    evidence_job.set_sub_file(fname)
if opts.condor_containerize_nonworker and singularity_image:
    evidence_job.add_condor_cmd("+SingularityImage", '"' + singularity_image + '"')
_add_hpip_condor_env(evidence_job)
evidence_job.write_sub_file()

alias_grid_job = None
if _use_hpip_pipeline:
    alias_grid_script = opts.working_directory + "/alias_hyperpipeline_grid.sh"
    with open(alias_grid_script, "w") as f:
        f.write("""#!/usr/bin/env bash
set -e
iter="$1"
target="overlap-grid-${iter}.dat"
alias="posterior_samples-${iter}.dat"
if [ ! -e "${target}" ]; then
  echo "alias_hyperpipeline_grid.sh: missing ${target}" >&2
  exit 1
fi
ln -sf "${target}" "${alias}"
""")
    os.chmod(alias_grid_script, os.stat(alias_grid_script).st_mode | stat.S_IEXEC)
    alias_grid_job, alias_grid_job_name = dag_utils.write_convert_sub(
        tag='alias_hpip_grid', exe=alias_grid_script, log_dir='',
        file_input='$(macroiteration)', file_output='/dev/null',
        arg_str='', out_dir=opts.working_directory, universe=local_worker_universe,
        no_grid=no_worker_grid)
    alias_grid_job.add_condor_cmd("initialdir", opts.working_directory)
    alias_grid_job.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/alias-hpip-grid-$(cluster)-$(process).log")
    alias_grid_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/alias-hpip-grid-$(cluster)-$(process).err")
    alias_grid_job.add_condor_cmd('request_disk', opts.general_request_disk)
    if opts.use_full_submit_paths:
        fname = opts.working_directory+"/"+alias_grid_job.get_sub_file()
        alias_grid_job.set_sub_file(fname)
    if opts.condor_containerize_nonworker and singularity_image:
        alias_grid_job.add_condor_cmd("+SingularityImage", '"' + singularity_image + '"')
    _add_hpip_condor_env(alias_grid_job)
    alias_grid_job.write_sub_file()


##   Fit job: default case
cip_args_base = cip_args
cip_request_disk=opts.general_request_disk
if opts.cip_request_disk:
    cip_request_disk = opts.cip_request_disk
#out_dir_base = opts.working_directory
cip_exe =opts.cip_exe
cip_exe_master = None
if opts.cip_exe:
    cip_exe_master = "{}".format(cip_exe) # force clpy, so NOT BY REFERENCE
cip_no_grid = no_worker_grid
if opts.use_osg_cip:
    cip_no_grid = False
cip_initialdir = opts.working_directory
cip_transfer_files = ['all.net']
if opts.cip_explode_jobs:
    cip_initialdir = opts.working_directory+"/iteration_$(macroiteration)_cip"
    cip_transfer_files = ['../all.net']
if (opts.cip_explode_jobs):
    if not opts.cip_explode_jobs_flat:
        cip_args_base += " --fit-save-gp " + out_dir_inside_cip + "/my_fit"
    else:
        cip_exe_master = "/bin/true"
#    out_dir_base += "/iteration_$(macroiteration)_cip/"
cip_job, cip_job_name = dag_utils.write_CIP_sub(tag='CIP',log_dir=None,arg_str=cip_args_base,request_memory=opts.request_memory_CIP,input_net=working_dir_inside_cip+'/all.net',output='overlap-grid-$(macroiterationnext)',out_dir=out_dir_inside_cip,exe=cip_exe_master,universe=local_worker_universe_cip,no_grid=cip_no_grid,use_osg=opts.use_osg_cip,use_singularity=opts.use_osg_cip and opts.use_singularity,singularity_image=singularity_image,use_simple_osg_requirements=opts.use_osg_cip,use_oauth_files=opts.use_oauth_files,transfer_files=cip_transfer_files)
# Modify: set 'initialdir'
cip_job.add_condor_cmd("initialdir",cip_initialdir)
# Modify output argument: change logs and working directory to be subdirectory for the run
cip_job.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/cip-$(cluster)-$(process).log")
cip_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/cip-$(cluster)-$(process).err")
cip_job.set_stdout_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/cip-$(cluster)-$(process).out")
cip_job.add_condor_cmd('request_disk',opts.cip_request_disk)
if opts.cip_explode_jobs:
    cip_job.add_arg('--not-worker')  # prevent anything that causes failure, reduce writing of output
if opts.use_full_submit_paths:
    fname = opts.working_directory+"/"+cip_job.get_sub_file()
    cip_job.set_sub_file(fname)
_add_hpip_condor_env(cip_job)
cip_job.write_sub_file()
if opts.cip_explode_jobs:
    print(" Exploding stage 1, with  ",opts.cip_explode_jobs, " workers producing samples ")
    cip_args_base = cip_args_base.replace('fit-save-gp','fit-load-gp')
    n_explode =opts.cip_explode_jobs
    fname_out = 'overlap-grid-$(macroiterationnext)-$(process)'
    if opts.cip_explode_jobs_dag:
        n_explode=1
        fname_out = 'overlap-grid-$(macroiterationnext)-$(cluster)'
    cip_args_base = cip_args_base.replace('my_fit', 'my_fit.pkl')  # yes, asymmetric arguments
    transfer_files_cip =['../all.net']
    if opts.use_osg_cip and 'fit-method gp' in cip_args_base:
        transfer_files_cip += ['my_fit.pkl']  # transfer current working directory fit
    cip_job_worker, cip_job_worker_name = dag_utils.write_CIP_sub(tag='CIP_worker',log_dir=None,arg_str=cip_args_base,request_memory=opts.request_memory_CIP,request_memory_flex=opts.request_memory_CIP_flex,input_net=working_dir_inside_cip+'/all.net',output=fname_out,out_dir=out_dir_inside_cip,exe=cip_exe,ncopies=n_explode,universe=local_worker_universe_cip,no_grid=cip_no_grid,use_osg=opts.use_osg_cip,use_singularity=opts.use_osg_cip and opts.use_singularity,singularity_image=singularity_image,use_simple_osg_requirements=opts.use_osg_cip,use_oauth_files=opts.use_oauth_files,transfer_files=transfer_files_cip)
    # Modify: set 'initialdir'
    cip_job_worker.add_condor_cmd("initialdir",opts.working_directory+"/iteration_$(macroiteration)_cip")
    # Modify output argument: change logs and working directory to be subdirectory for the run
    cip_job_worker.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/cip-$(cluster)-$(process).log")
    cip_job_worker.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/cip-$(cluster)-$(process).err")
    cip_job_worker.set_stdout_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/cip-$(cluster)-$(process).out")
    cip_job_worker.add_condor_cmd('request_disk',cip_request_disk)
    if opts.use_full_submit_paths:
        fname = opts.working_directory+"/"+cip_job_worker.get_sub_file()
        cip_job_worker.set_sub_file(fname)
    _add_hpip_condor_env(cip_job_worker)
    cip_job_worker.write_sub_file()

    # Create worker join job
    #   ... if we are using cip_explode_jobs_dag, we need a different script entirely
    #     filename to join : note *two* minus signs, because we don't want the non-worker job to match! It has tiny output. This can cause pathological behavior for the random join stage
    join_cip_job,join_cip_job_name = dag_utils.write_joingrids_sub(input_pattern=opts.working_directory+"/iteration_$(macroiteration)_cip/output-grid-*-*.{}".format(grid_suffix),target_dir=opts.working_directory,output_base="overlap-grid-$(macroiterationnext)",n_explode=n_explode,log_dir=opts.working_directory+"/iteration_$(macroiteration)_cip/logs",universe=local_worker_universe,no_grid=no_worker_grid,extra_text=extra_text)
    join_cip_job.add_condor_cmd("initialdir",opts.working_directory+"/iteration_$(macroiteration)_cip")
    # Modify output argument: change logs and working directory to be subdirectory for the run
    join_cip_job.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/join-$(cluster)-$(process).log")
    join_cip_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/join-$(cluster)-$(process).err")
    join_cip_job.set_stdout_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/join-$(cluster)-$(process).out")
    join_cip_job.add_condor_cmd('request_disk',opts.general_request_disk)
    if opts.use_full_submit_paths:
        fname = opts.working_directory+"/"+join_cip_job.get_sub_file()
        join_cip_job.set_sub_file(fname)
    if opts.condor_containerize_nonworker and singularity_image:
        join_cip_job.add_condor_cmd("+SingularityImage", '"' + singularity_image + '"')
    _add_hpip_condor_env(join_cip_job)
    join_cip_job.write_sub_file()

##   puffball job: default case
if puff_args and puff_cadence:
    puff_job, puff_job_name = dag_utils.write_puff_sub(tag='PUFF',log_dir=None,arg_str=puff_args,request_memory=opts.request_memory_ILE,input_net=opts.working_directory+'/overlap-grid-$(macroiterationnext).{}'.format(grid_suffix),output=opts.working_directory+'/puffball-$(macroiterationnext)',out_dir=opts.working_directory,exe=opts.puff_exe,universe=local_worker_universe,no_grid=no_worker_grid,extra_text=extra_text)
    # Modify: set 'initialdir' to CIP WORKING DIR 
    puff_job.add_condor_cmd("initialdir",opts.working_directory+"/iteration_$(macroiteration)_cip")
    # Modify output argument: change logs and working directory to be subdirectory for the run
    puff_job.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/puff-$(cluster)-$(process).log")
    puff_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/puff-$(cluster)-$(process).err")
    puff_job.set_stdout_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/puff-$(cluster)-$(process).out")
    puff_job.add_condor_cmd('request_disk',opts.general_request_disk)
    if opts.use_full_submit_paths:
        fname = opts.working_directory+"/"+puff_job.get_sub_file()
        puff_job.set_sub_file(fname)
    if opts.condor_containerize_nonworker and singularity_image:
        puff_job.add_condor_cmd("+SingularityImage", '"' + singularity_image + '"')
    _add_hpip_condor_env(puff_job)
    puff_job.write_sub_file()

# LISA reflected-sky-mode: a per-iteration node that reflects the CIP-proposed
# grid's sky to the secondary mode (convert_primary_sky_mode_to_secondary),
# overwriting overlap-grid-(it+1) in place so the NEXT iteration's ILE explores
# the reflected mode.  arg_str ends with "--fname " so write_convert_sub's
# add_arg(file_input) becomes --fname <grid>.
reflected_mode_job = None
reflected_mode_iter = None
if opts.search_reflected_sky_mode and int(opts.n_iterations) > 1:
    reflect_exe = opts.reflected_sky_mode_exe or dag_utils.which("convert_primary_sky_mode_to_secondary")
    reflected_mode_iter = opts.search_reflected_sky_mode_iteration
    if reflected_mode_iter is None:
        reflected_mode_iter = int(opts.n_iterations) - 2
    if reflected_mode_iter <= 0:
        reflected_mode_iter = int(opts.n_iterations) - 1
    reflected_mode_job, reflected_mode_job_name = dag_utils.write_convert_sub(
        tag='reflected_sky_mode', log_dir=None,
        arg_str="--lisa-reference-time {} --fname ".format(opts.lisa_reference_time),
        file_input=opts.working_directory+'/overlap-grid-$(macroiterationnext).{}'.format(grid_suffix),
        file_output=opts.working_directory+'/iteration_$(macroiteration)_cip/logs/reflected-sky-$(cluster)-$(process).out',
        out_dir=opts.working_directory, exe=reflect_exe,
        universe=local_worker_universe, no_grid=no_worker_grid)
    reflected_mode_job.add_condor_cmd("initialdir", opts.working_directory+"/iteration_$(macroiteration)_cip")
    reflected_mode_job.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/reflected-sky-$(cluster)-$(process).log")
    reflected_mode_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/reflected-sky-$(cluster)-$(process).err")
    reflected_mode_job.add_condor_cmd('request_disk', opts.general_request_disk)
    if opts.use_full_submit_paths:
        reflected_mode_job.set_sub_file(opts.working_directory+"/"+reflected_mode_job.get_sub_file())
    if opts.condor_containerize_nonworker and singularity_image:
        reflected_mode_job.add_condor_cmd("+SingularityImage", '"' + singularity_image + '"')
    _add_hpip_condor_env(reflected_mode_job)
    reflected_mode_job.write_sub_file()


##   calibration PILOT job (Option C adaptive cal; see RIFT/calmarg/DESIGN_adaptive_driver.md)
calpilot_job = None
calpilot_node_per_iteration = {}
if opts.calmarg_pilot:
    calpilot_log_dir = opts.working_directory + "/iteration_$(macroiteration)_cip/logs/"
    # On OSG the CALPILOT job runs ILE internally, so it needs the same container + input
    # transfer set as the wide ILE jobs (PSD + cal envelopes + frames), plus the composite.
    # transfer_file_names[:-1] = the helper transfer list WITHOUT the wide grid (the pilot
    # harvests its own grid from the transferred composite).
    calpilot_job, calpilot_job_name = dag_utils.write_calpilot_sub(
        tag='CALPILOT', log_dir=calpilot_log_dir, working_directory=opts.working_directory,
        ile_args_file=opts.working_directory + "/args_ile.txt",
        top_fraction=opts.calmarg_pilot_top_fraction, max_points=opts.calmarg_pilot_max_points,
        request_memory=opts.request_memory_ILE, request_gpu=opts.request_gpu_ILE,
        universe="vanilla",
        singularity_image=(singularity_image if (opts.use_singularity or opts.use_osg or opts.condor_containerize_nonworker) else None),
        use_osg=opts.use_osg, use_singularity=opts.use_singularity, frames_dir=opts.frames_dir,
        use_oauth_files=opts.use_oauth_files,
        transfer_files=(list(transfer_file_names_pilot_base) if opts.use_osg else None))
    calpilot_job.add_condor_cmd("initialdir", opts.working_directory)
    calpilot_job.add_condor_cmd('request_disk',opts.general_request_disk)
    if opts.use_full_submit_paths:
        calpilot_job.set_sub_file(opts.working_directory + "/" + calpilot_job.get_sub_file())
    calpilot_job.write_sub_file()


##   EXTRINSIC handoff consolidation job (see RIFT/calmarg/DESIGN_extrinsic_handoff.md).
##   Pure-python LOCAL-universe job: picks the most representative per-event extrinsic
##   proposal of iteration $(macroiteration) -> extr_consolidated_<it>.npz, seeding wide_{N+1}.
extrconsolidate_job = None
extrconsolidate_node_per_iteration = {}
if opts.extrinsic_handoff:
    extrconsolidate_log_dir = opts.working_directory + "/iteration_$(macroiteration)_ile/logs/"
    extrconsolidate_job, extrconsolidate_job_name = dag_utils.write_extrconsolidate_sub(
        tag='EXTRCONSOLIDATE', log_dir=extrconsolidate_log_dir,
        working_directory=opts.working_directory, select=opts.extrinsic_handoff_select)
    extrconsolidate_job.add_condor_cmd("initialdir", opts.working_directory)
    if opts.use_full_submit_paths:
        extrconsolidate_job.set_sub_file(opts.working_directory + "/" + extrconsolidate_job.get_sub_file())
    extrconsolidate_job.write_sub_file()


##   fetch job.  Fetch is before CURRENT iteration, so it is just-in-time and can be used at iteration 0
if fetch_args:
    fetch_job, fetch_job_name = dag_utils.write_puff_sub(tag='FETCH',log_dir=None,arg_str=fetch_args,request_memory=opts.request_memory_ILE,input_net=None,output=opts.working_directory+'/fetch-$(macroiteration).{}'.format(grid_suffix),out_dir=opts.working_directory,exe=opts.fetch_ext_grid_exe,universe=local_worker_universe,no_grid=no_worker_grid,extra_text=extra_text)
    # Modify: set 'initialdir' to CIP WORKING DIR 
    fetch_job.add_condor_cmd("initialdir",opts.working_directory+"/iteration_$(macroiteration)_cip")
    # Modify output argument: change logs and working directory to be subdirectory for the run
    fetch_job.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/fetch-$(cluster)-$(process).log")
    fetch_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/fetch-$(cluster)-$(process).err")
    fetch_job.set_stdout_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/fetch-$(cluster)-$(process).out")
    fetch_job.add_condor_cmd('request_disk',opts.general_request_disk)
    if opts.use_full_submit_paths:
        fname = opts.working_directory+"/"+fetch_job.get_sub_file()
        fetch_job.set_sub_file(fname)
    if opts.condor_containerize_nonworker and singularity_image:
        fetch_job.add_condor_cmd("+SingularityImage", '"' + singularity_image + '"')
    _add_hpip_condor_env(fetch_job)
    fetch_job.write_sub_file()



cip_job_list = None
if  (cip_args_lines is None):
    # Write the default cip_job into cip_job_list n times
    # add some redundancy in case this happens : we're hitting a problem with having too few copies of this/fencepost error later
    if opts.cip_explode_jobs is None:
        cip_job_list =(opts.n_iterations)* [cip_job ]
    else:
        cip_job_list =(opts.n_iterations)* [ [cip_job, cip_job_worker] ]
else:  
    # we have different cip jobs for different iteration numbers
    cip_job_list = []
    for indx in np.arange(len(cip_args_lines)):
#        out_dir_base = opts.working_directory
        cip_args_extra = ""
        cip_args_truncate=''
        # If we are a 'G' iteration, flag to use other executable
        cip_exe_here = None
        cip_exe_here_master = None
        if cip_exe:
            cip_exe_here = "{} ".format(cip_exe)
            cip_exe_here_master = " {} ".format(cip_exe_master)
        if indx in creating_G_list:
            cip_exe_here = opts.cip_exe_G
        if opts.cip_explode_jobs:
            if not opts.cip_explode_jobs_flat:
                cip_args_extra += " --fit-save-gp " + out_dir_inside_cip+"/my_fit"
            # else:
            #     cip_exe_here_master = "/bin/true"
#            out_dir_base += "/iteration_$(macroiteration)_cip/"
            # set n_eff for primary job to be small. ONLY used for the primary non-worker job
            cip_args_truncate = " --n-eff 5 --n-max 10000 --n-output-samples 1 "  # cap n_eff, number of iterations, and *output samples* for non-worker jobs, to avoid contamination
        # Write the appropriate CIP jobs. [note only one CIP per iteration, so unique
#        cip_exe_here =cip_exe
        cip_job, cip_job_name = dag_utils.write_CIP_sub(tag='CIP_'+str(indx),log_dir=None,arg_str=cip_args_lines[indx]+cip_args_extra+cip_args_truncate,request_memory=opts.request_memory_CIP,request_memory_flex=opts.request_memory_CIP_flex,input_net=working_dir_inside_cip+'/all.net',output='overlap-grid-$(macroiterationnext)',out_dir=out_dir_inside_cip,exe=cip_exe_here_master,universe=local_worker_universe_cip,no_grid=cip_no_grid,use_osg=opts.use_osg_cip,use_singularity=opts.use_osg_cip and opts.use_singularity,singularity_image=singularity_image,use_simple_osg_requirements=opts.use_osg_cip,use_oauth_files=opts.use_oauth_files,transfer_files=cip_transfer_files)
        # Modify: set 'initialdir'
        cip_job.add_condor_cmd("initialdir",cip_initialdir)
        # Modify output argument: change logs and working directory to be subdirectory for the run
        cip_job.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/cip-$(cluster)-$(process).log")
        cip_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/cip-$(cluster)-$(process).err")
        cip_job.set_stdout_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/cip-$(cluster)-$(process).out")
        cip_job.add_condor_cmd('request_disk',cip_request_disk)
        if opts.cip_explode_jobs:
            cip_job.add_arg("--not-worker")
        if opts.use_full_submit_paths:
            fname = opts.working_directory+"/"+cip_job.get_sub_file()
            cip_job.set_sub_file(fname)
        _add_hpip_condor_env(cip_job)
        cip_job.write_sub_file()


        if opts.cip_explode_jobs:
           print(" Exploding stage 2, with  ",opts.cip_explode_jobs, " workers producing samples ")
           n_explode =opts.cip_explode_jobs
           fname_out = 'overlap-grid-$(macroiterationnext)-$(process)'
           if opts.cip_explode_jobs_dag:
                n_explode=1
                fname_out = 'overlap-grid-$(macroiterationnext)-$(cluster)'
           cip_args_extra = cip_args_extra.replace('fit-save-gp','fit-load-gp')
           cip_args_extra = cip_args_extra.replace('my_fit','my_fit.pkl')
           # if the last set of workers, we are making extrinsic samples, so change the output samples  
           if indx == len(cip_args_lines) -1 and opts.last_iteration_extrinsic: # on last iteration, doing extrinsic phase
               n_samples_per_job = int(opts.last_iteration_extrinsic_nsamples/(1.0*opts.cip_explode_jobs_last))
               cip_args_extra +=" --n-eff {} --n-output-samples {} ".format(n_samples_per_job,n_samples_per_job)
               if not (opts.use_eccentricity_squared_sampling):
                   cip_args_lines[indx] = cip_args_lines[indx].replace(' --parameter eccentricity_squared ',' --parameter-implied eccentricity_squared --parameter-nofit eccentricity ')
           transfer_files_cip=['../all.net']
           if opts.use_osg_cip and 'fit-method gp' in cip_args_base:
               transfer_files_cip += ['my_fit.pkl']  # transfer current working directory fit
           cip_job_worker, cip_job_worker_name = dag_utils.write_CIP_sub(tag='CIP_worker'+str(indx),log_dir=None,arg_str=cip_args_lines[indx]+cip_args_extra,request_memory=opts.request_memory_CIP,request_memory_flex=opts.request_memory_CIP_flex,input_net=working_dir_inside_cip+'/all.net',output=fname_out,out_dir=out_dir_inside_cip,exe=cip_exe_here,ncopies=n_explode,universe=local_worker_universe_cip,no_grid=cip_no_grid,use_osg=opts.use_osg_cip,use_singularity=opts.use_osg_cip and opts.use_singularity,singularity_image=singularity_image,use_simple_osg_requirements=opts.use_osg_cip,use_oauth_files=opts.use_oauth_files,transfer_files=transfer_files_cip)
           # Modify: set 'initialdir'
           cip_job_worker.add_condor_cmd("initialdir",opts.working_directory+"/iteration_$(macroiteration)_cip")
           # Modify output argument: change logs and working directory to be subdirectory for the run
           cip_job_worker.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/cipworker-$(cluster)-$(process).log")
           cip_job_worker.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/cipworker-$(cluster)-$(process).err")
           cip_job_worker.set_stdout_file(opts.working_directory+"/iteration_$(macroiteration)_cip/logs/cipworker-$(cluster)-$(process).out")
           cip_job_worker.add_condor_cmd('request_disk',cip_request_disk)
           if opts.use_full_submit_paths:
               fname = opts.working_directory+"/"+cip_job_worker.get_sub_file()
               cip_job_worker.set_sub_file(fname)
           _add_hpip_condor_env(cip_job_worker)
           cip_job_worker.write_sub_file()


        # augment the CIP job list
        # Note, if we have a subdag, we create the word "subdag"
        if not(cip_args_prefixes[indx][0]=='Z'):
            n_to_add = int(cip_args_n[indx])
            if (indx == len(cip_args_lines)-1) and np.sum(cip_args_n[:indx])<opts.n_iterations:
                n_to_add = int(opts.n_iterations - np.sum(cip_args_n[:(indx-1)]) )
            if opts.cip_explode_jobs is None:
                cip_job_list = cip_job_list + n_to_add*[cip_job]
            else:
                print(" Exploding stage 3 ")
                cip_job_list = cip_job_list + n_to_add*[[cip_job,cip_job_worker]]
        else:
            print(" Indefinite subdag-to-convergence requested for group ", indx)
            cip_job_list += ["Subdag"]
print(" ===> Iteration size ", len(cip_job_list), opts.n_iterations)
##   Test job (terminate, convergence
if opts.test_args:
    test_node_list = []
    ##  Convert job : make results accessible after every iteration.  (Only performed if the tests are active, to make my life easier)
    # Hyperpipeline mode: the input grid is already .dat (hyperpipeline
    # format), while this convert step targets LI posterior_samples.dat,
    # which is a different schema (mass_1, spin1x, ...).  Skip emitting the
    # convert job and point the convergence test directly at the grid files.
    if not _use_hpip_pipeline:
        convert_job, convert_job_name =dag_utils.write_convert_sub(tag='convert',log_dir=None,arg_str=convert_args,file_input=opts.working_directory+'/overlap-grid-$(macroiteration).xml.gz', file_output=opts.working_directory+'/posterior_samples-$(macroiteration).dat' ,out_dir=opts.working_directory,exe=opts.test_exe,universe=local_worker_universe,no_grid=no_worker_grid)
        convert_job.add_condor_cmd("initialdir",opts.working_directory)
        convert_job.set_log_file(opts.working_directory+"/iteration_$(macroiterationlast)_test/logs/convert-$(cluster)-$(process).log")
        convert_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiterationlast)_test/logs/convert-$(cluster)-$(process).err")
        convert_job.add_condor_cmd('request_disk',opts.general_request_disk)
        if opts.use_full_submit_paths:
            fname = opts.working_directory+"/"+convert_job.get_sub_file()
            convert_job.set_sub_file(fname)
        if opts.condor_containerize_nonworker and singularity_image:
            convert_job.add_condor_cmd("+SingularityImage", '"' + singularity_image + '"')
        convert_job.write_sub_file()


    _test_samples_files = [
        opts.working_directory+'/posterior_samples-$(macroiteration).dat',
        opts.working_directory+'/posterior_samples-$(macroiterationlast).dat']
    test_job, test_job_name = dag_utils.write_test_sub(tag='test',log_dir=None,arg_str=test_args,samples_files=_test_samples_files ,out_dir=opts.working_directory,exe=opts.test_exe,universe=local_worker_universe,no_grid=no_worker_grid)
    # Modify: set 'initialdir'
    test_job.add_condor_cmd("initialdir",opts.working_directory+"/iteration_$(macroiteration)_test")
    # Modify output argument: change logs and working directory to be subdirectory for the run
    test_job.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_test/logs/test-$(cluster)-$(process).log")
    test_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_test/logs/test-$(cluster)-$(process).err")
    test_job.set_stdout_file(opts.working_directory+"/iteration_$(macroiteration)_test/logs/test-$(cluster)-$(process).out")
    test_job.add_condor_cmd('request_disk',opts.general_request_disk)
    if opts.use_full_submit_paths:
        fname = opts.working_directory+"/"+test_job.get_sub_file()
        test_job.set_sub_file(fname)
    if opts.condor_containerize_nonworker and singularity_image:
        test_job.add_condor_cmd("+SingularityImage", '"' + singularity_image + '"')
    _add_hpip_condor_env(test_job)
    test_job.write_sub_file()


if opts.plot_args and opts.test_args:
    # default: last two iterations
    #  ... unless there is extrinsic samples, then use those
    #  ... note this has to work with PEsummary as well, which only takes one set of samples
#    samples_files =['posterior_samples-$(macroiteration).dat','posterior_samples-$(macroiterationlast).dat'] 
#    if opts.last_iteration_extrinsic:
#        samples_files =[]
    # User will be responsible for passing argument strings.  We are not hardcoding in the argument format
    samples_files =[]
    plot_job, plot_job_name = dag_utils.write_plot_sub(tag='plot',log_dir=None,arg_str=plot_args,samples_files=samples_files,out_dir=opts.working_directory,exe=opts.plot_exe,universe=local_worker_universe)
    plot_job.set_log_file(opts.working_directory+"/iteration_$(macroiteration)_plot/logs/test-$(cluster)-$(process).log")
    plot_job.set_stderr_file(opts.working_directory+"/iteration_$(macroiteration)_plot/logs/test-$(cluster)-$(process).err")
    plot_job.set_stdout_file(opts.working_directory+"/iteration_$(macroiteration)_plot/logs/test-$(cluster)-$(process).out")
    plot_job.add_condor_cmd('request_disk',opts.general_request_disk)
    if opts.use_full_submit_paths:
        fname = opts.working_directory+"/"+plot_job.get_sub_file()
        plot_job.set_sub_file(fname)
    plot_job.write_sub_file()


if opts.gridinit_args:
    gridinit_job, gridinit_job_name = dag_utils.write_init_sub(tag='grid',log_dir=None,arg_str=gridinit_args,out_dir=opts.working_directory,exe=opts.gridinit_exe,universe=local_worker_universe)
    gridinit_job.set_log_file(opts.working_directory+"/init-$(cluster)-$(process).log")
    gridinit_job.set_stderr_file(opts.working_directory+"/init--$(cluster)-$(process).err")
    gridinit_job.set_stdout_file(opts.working_directory+"/init-$(cluster)-$(process).out")
    gridinit_job.add_condor_cmd('request_disk',opts.general_request_disk)
    if opts.use_full_submit_paths:
        fname = opts.working_directory+"/"+gridinit_job.get_sub_file()
        gridinit_job.set_sub_file(fname)
    gridinit_job.write_sub_file()

if opts.frame_rotation:
    # Wriite command to do the conversion
    extra_arg=''
    if 'fref' in ile_args:
        # retrieve fref
        tmp_args = ile_args.split('--')
        lines = [x for x in tmp_args if x.startswith('reference-freq')]
        fref_str = lines[-1].split()[-1]
        extra_arg = ' --fref {} '.format(fref_str)
    with open('fix_frame_rot.sh','w') as f:
        f.write("""#!/bin/bash
if [ ! -e extrinsic_posterior_samples_orig.dat ]; then
   mv extrinsic_posterior_samples.dat  extrinsic_posterior_samples_orig.dat;
   convert_ascii_framechange_xphm.py --extrinsic-posterior-file   extrinsic_posterior_samples_orig .dat --fname-out extrinsic_posterior_samples.dat {}
fi;
""".format(extra_arg))
    # Write submit file for the conversion
    rotate_job, rotate_job_name = dag_utils.write_convert_sub(exe=which('fix_frame_rot.sh'),tag='ROTATE',log_dir=None,arg_str='',file_input='',file_output=None, out_dir=opts.working_directory,universe=local_worker_universe,no_grid=no_worker_grid)
    rotate_job.add_condor_cmd("initialdir",opts.working_directory)
    rotate_job.set_log_file(opts.working_directory+"/reweight-merge-$(cluster)-$(process).log")
    rotate_job.set_stderr_file(opts.working_directory+"/reweight-merge-$(cluster)-$(process).err")
    rotate_job.set_stdout_file(opts.working_directory+"/reweight-merge-$(cluster)-$(process).out")
    rotate_job.add_condor_cmd('request_disk',opts.cal_request_disk)
    if opts.use_full_submit_paths:
        fname = opts.working_directory+"/"+rotate_job.get_sub_file()
        rotate_job.set_sub_file(fname)
    if opts.use_osg:
        rotate_job.add_condor_cmd("+DESIRED_SITES",'"nogrid"')
        rotate_job.add_condor_cmd("+flock_local",'true')
    rotate_job.write_sub_file()

if opts.calibration_reweighting:
    # bilby ini must be created FIRST for correct operation  if writing cal envelopes
    if opts.bilby_ini_file:
        pickle_job, pickle_job_name = dag_utils.write_bilby_pickle_sub(tag='Bilby_pickle',log_dir=None,bilby_ini_file=opts.bilby_ini_file,exe=opts.bilby_pickle_exe,universe='local',no_grid=no_worker_grid,cache_file=opts.cache_file,frames_dir=opts.frames_dir,ile_args=ile_args)
        pickle_job.set_log_file(opts.working_directory+"/pickle-$(cluster)-$(process).log")
        pickle_job.set_stderr_file(opts.working_directory+"/pickle-$(cluster)-$(process).err")
        pickle_job.set_stdout_file(opts.working_directory+"/pickle-$(cluster)-$(process).out")
        pickle_job.add_condor_cmd('request_disk',opts.general_request_disk)

        if opts.use_full_submit_paths:
            fname = opts.working_directory+"/"+pickle_job.get_sub_file()
            pickle_job.set_sub_file(fname)
        pickle_job.write_sub_file()
    
    extra_args_dict = {}
    extra_args_dict['time_marg'] = not (opts.last_iteration_extrinsic)   # pass time_marg=False if we are resampling in time
    if opts.bilby_pickle_file:
        extra_args_dict['pickle_file'] = opts.bilby_pickle_file
    else:
        extra_args_dict['pickle_file'] =opts.working_directory+"/calmarg/data/calmarg_data_dump.pickle"
    # OSG options
    if opts.calibration_reweighting_osg:
        extra_args_dict['use_osg'] =  opts.use_osg
        extra_args_dict['use_singularity'] =  opts.use_singularity
        extra_args_dict['singularity_image'] = singularity_image
        extra_args_dict['universe'] = 'vanilla'
    else:
        extra_args_dict['no_grid']  = no_worker_grid
        extra_args_dict['universe'] = local_worker_universe
    if opts.transfer_file_list:
        # Handle scenario where h5 file needed by calmarg job. Note arranged as transfer because we don't know if execute nodes have the file, even if local
        any_h5_needed_list = [x for x in transfer_file_names if x.endswith('.h5')]  # NRSur, etc h5 files we need for lalsuite
        extra_args_dict['transfer_files'] = any_h5_needed_list
        
    calibration_job, calibration_job_name = dag_utils.write_calibration_uncertainty_reweighting_sub(tag='Calib_reweight',log_dir=None,posterior_file=opts.working_directory+"/extrinsic_posterior_samples.dat",exe=opts.calibration_reweighting_exe,ile_args=ile_args,n_cal=opts.calibration_reweighting_count,use_oauth_files=opts.use_oauth_files,**extra_args_dict)
        
    if opts.calibration_reweighting_initial_extra_args:
        # properly handle quoted extra arguments, such as dictionaries of waveform arguments
        add_str = opts.calibration_reweighting_initial_extra_args
        if '"' in add_str:
            add_str = dag_utils.safely_quote_arg_str(add_str)
        calibration_job.add_arg(add_str)
    if not(opts.calibration_reweighting_batchsize): # single run
        calibration_job.set_log_file(opts.working_directory+"/calmarg/logs/calibration-$(cluster)-$(process).log")
        calibration_job.set_stderr_file(opts.working_directory+"/calmarg/logs/calibration-$(cluster)-$(process).err")
        calibration_job.set_stdout_file(opts.working_directory+"/calmarg/logs/calibration-$(cluster)-$(process).out")
    else:
        if not(opts.calibration_reweighting_osg):
            calibration_job.add_condor_cmd("initialdir",opts.working_directory+"/calmarg/")
        else:
            calibration_job.add_condor_cmd("initialdir",opts.working_directory)            
        calibration_job.add_arg(" --start_index $(macrostartidx) ")
        calibration_job.add_arg(" --end_index $(macroendidx) ")
        calibration_job.set_log_file(opts.working_directory+"/calmarg/logs/calibration-$(macrostartidx)-$(cluster)-$(process).log")
        calibration_job.set_stderr_file(opts.working_directory+"/calmarg/logs/calibration-$(macrostartidx)-$(cluster)-$(process).err")
        calibration_job.set_stdout_file(opts.working_directory+"/calmarg/logs/calibration-$(macrostartidx)-$(cluster)-$(process).out")
    if opts.use_singularity and opts.calibration_reweighting_osg:
        calibration_job.add_condor_cmd('request_disk',opts.ile_request_disk)  # similar task, so use similar request
    else:
        calibration_job.add_condor_cmd('request_disk',opts.general_request_disk)

    if opts.use_full_submit_paths:
        fname = opts.working_directory+"/"+calibration_job.get_sub_file()
        calibration_job.set_sub_file(fname)
    calibration_job.write_sub_file()

    # job to combine things. Write here to make logic human-readable
    if (opts.calibration_reweighting_batchsize): # single run
        reweight_merge_job, reweight_merge_job_name = dag_utils.write_convert_sub(exe=which('combine_weights_and_rejection_sample.py'),tag='CAL_REWEIGHT_COMBINE',log_dir=None,arg_str=opts.calibration_reweighting_extra_args,file_input=opts.working_directory+'/extrinsic_posterior_samples.dat weight_files/',file_output=None, out_dir=opts.working_directory,universe=local_worker_universe,no_grid=no_worker_grid)
        reweight_merge_job.add_condor_cmd("initialdir",opts.working_directory) # NOT calmarg, because the normal 'weight files' is put in the location of the *samples*, which is the rundir
        reweight_merge_job.set_log_file(opts.working_directory+"/calmarg/logs/reweight-merge-$(cluster)-$(process).log")
        reweight_merge_job.set_stderr_file(opts.working_directory+"/calmarg/logs/reweight-merge-$(cluster)-$(process).err")
        reweight_merge_job.set_stdout_file(opts.working_directory+"/calmarg/logs/reweight-merge-$(cluster)-$(process).out")
        reweight_merge_job.add_condor_cmd('request_disk',opts.cal_request_disk)
        if opts.use_full_submit_paths:
            fname = opts.working_directory+"/"+reweight_merge_job.get_sub_file()
            reweight_merge_job.set_sub_file(fname)
        if opts.use_osg:
            reweight_merge_job.add_condor_cmd("+DESIRED_SITES",'"nogrid"')
            reweight_merge_job.add_condor_cmd("+flock_local",'true')
        reweight_merge_job.write_sub_file()



    

if opts.comov_distance_reweighting:
    if opts.calibration_reweighting:
        print(opts.convert_ascii2h5_exe)
        convert_ascii_job, convert_ascii_job_name = dag_utils.write_convert_ascii_to_h5_sub(tag='Convert_ascii2h5',log_dir=None,output_file=opts.working_directory+"/posterior_samples.h5",posterior_file=opts.working_directory+"/reweighted_posterior_samples.dat",convert_ascii_to_h5_exe=opts.convert_ascii2h5_exe,universe=local_worker_universe,no_grid=no_worker_grid)
        convert_ascii_job.set_log_file(opts.working_directory+"/convert-ascii-$(cluster)-$(process).log")
        convert_ascii_job.set_stderr_file(opts.working_directory+"/convert_ascii-$(cluster)-$(process).err")
        convert_ascii_job.set_stdout_file(opts.working_directory+"/convert_ascii-$(cluster)-$(process).out")
        convert_ascii_job.add_condor_cmd('request_disk',opts.general_request_disk)

        if opts.use_full_submit_paths:
            fname = opts.working_directory+"/"+convert_ascii_job.get_sub_file()
            convert_ascii_job.set_sub_file(fname)
        convert_ascii_job.write_sub_file()
    else:
        convert_ascii_job, convert_ascii_job_name = dag_utils.write_convert_ascii_to_h5_sub(tag='Convert_ascii2h5',log_dir=None,output_file=opts.working_directory+"/posterior_samples.h5",posterior_file=opts.working_directory+"/extrinsic_posterior_samples.dat",convert_ascii_to_h5_exe=opts.convert_ascii2h5_exe,universe=local_worker_universe)
        convert_ascii_job.set_log_file(opts.working_directory+"/convert-ascii-$(cluster)-$(process).log")
        convert_ascii_job.set_stderr_file(opts.working_directory+"/convert_ascii-$(cluster)-$(process).err")
        convert_ascii_job.set_stdout_file(opts.working_directory+"/convert_ascii-$(cluster)-$(process).out")
        convert_ascii_job.add_condor_cmd('request_disk',opts.general_request_disk)

        if opts.use_full_submit_paths:
            fname = opts.working_directory+"/"+convert_ascii_job.get_sub_file()
            convert_ascii_job.set_sub_file(fname)
        convert_ascii_job.write_sub_file()

    
    distance_job, distance_job_name = dag_utils.write_comov_distance_reweighting_sub(tag='Comov_dist',log_dir=None,reweight_location=opts.working_directory+"/cosmo_reweight.h5",posterior_file=opts.working_directory+"/posterior_samples.h5",comov_distance_reweighting_exe=opts.comov_distance_reweighting_exe,universe=local_worker_universe)
    distance_job.set_log_file(opts.working_directory+"/pickle-$(cluster)-$(process).log")
    distance_job.set_stderr_file(opts.working_directory+"/pickle-$(cluster)-$(process).err")
    distance_job.set_stdout_file(opts.working_directory+"/pickle-$(cluster)-$(process).out")
    distance_job.add_condor_cmd('request_disk',opts.general_request_disk)

    if opts.use_full_submit_paths:
        fname = opts.working_directory+"/"+distance_job.get_sub_file()
        distance_job.set_sub_file(fname)
    distance_job.write_sub_file()
##   Convert job(s)
#    - relocate input grid to correct location.  (MAKE TRIVIAL: do via output arguments in previous stage!)
##   Assessment job (no-op for now)




# ++++
# Create workflow 
# ++++

# Create workflow
#   - Create grid node as needed
#   - Loop over iterations
#      - if iteration0, use seed grid (should already be copied in place)
#      - if not iteration 0, grid should be in place (from previous stage)
#      - Loop over events, make ILE node per event
#      - create consolidate job, make it depend on all events in that iteration
#      - create fit job, make it depend on consolidate job
#    

parent_fit_node = None
last_puff_node = None
last_node=None

if opts.gridinit_args:
   grid_node = pipeline.CondorDAGNode(gridinit_job) 
   dag.add_node(grid_node)
   parent_fit_node = grid_node  # this must happen before the first ILE jobs

convert_psd_node_list = []
if opts.use_bw_psd:
    print(" ===> Adding BW PSD nodes to dag <=== ")

#    bw1_node =pipeline.CondorDAGNode(bw2_job)
#    bw1_node.add_parent(bw0_node)
#    dag.add_node(bw1_node)

    # Add nodes to convert the PSD to the correct format
    # 
    for ifo in list(channel_names.keys()):
        bw0_node =pipeline.CondorDAGNode(bw_job)
        bw0_node.add_macro('ifo',ifo)
        # Create argument string
        bw_arg_str = ''
        channel_name, channel_flow = channel_dict_bw[ifo]
        bw_arg_str += " --ifo "+ifo
        bw_arg_str += " --"+ifo+"-channel "+ifo+":"+channel_name
        ifo_cache_name =  working_dir_inside_local +"/for_bw_"+ifo+".cache"  # bw requires single-IFO cache files, argh!!! Note these are NOT on OSG
        ifo_char = ifo[0]
        os.system("grep ^"+ifo_char +  " " +opts_ile.cache_file+ " > " + ifo_cache_name)
        bw_arg_str += " --"+ifo+"-cache "+ifo_cache_name
        bw_arg_str += " --"+ifo+"-flow "+str(channel_flow)
        bw_arg_str += " --"+ifo+"-timeslide 0.0"
        bw0_node.add_macro('macroargument0',bw_arg_str)
        bw0_node.set_retry(opts.ile_retries)  # these fail all the time
        dag.add_node(bw0_node)

        # Need correct inheritance: this will be annoying b/c loop below does not allow for it
        # ...so do the PSD convert steps *serially* for now.  (The individual PSDs can be done in parallel)
        convert_psd_node =pipeline.CondorDAGNode(convert_psd_job)
        convert_psd_node.add_parent(bw0_node)
#        if not(parent_fit_node is None):
#            convert_psd_node.add_parent(parent_fit_node)
        parent_fit_node = convert_psd_node
        convert_psd_node.add_macro('ifo',ifo)
        convert_psd_node.set_retry(opts.ile_retries)  # these fail all the time
        dag.add_node(convert_psd_node)
        convert_psd_node_list.append(convert_psd_node)

n_group = opts.ile_n_events_to_analyze
n_group_first = opts.ile_n_events_to_analyze
if opts.ile_n_events_to_analyze_first:
    n_group_first= opts.ile_n_events_to_analyze_first

con_node_list = []
unify_node_list = []
ile_node_list_per_iteration={}
ile_target_worklevel_per_iteration={}
cip_subdag_nodes = {}
cip_subdag_work_targets = {} 
for it in np.arange(it_start,opts.n_iterations):
    ile_node_list_per_iteration[it] = []
    print(it, opts.n_iterations)
    consolidate_now = None
    fit_node_now = None
    # Create consolidate job
    con_node = pipeline.CondorDAGNode(con_job)
    con_node.add_macro("macroiteration",it)
    con_node.set_retry(opts.general_retries)
    con_node_list.append(con_node)
    # Create unify job
    unify_node = pipeline.CondorDAGNode(unify_job)
    unify_node.add_macro("macroiteration",it)
    unify_node.add_parent(con_node)
    unify_node.set_retry(opts.general_retries)
    if not(it ==0) and not(opts.first_iteration_jumpstart):  # don't require first composite to be nonempty if we are running a jumpstart!
        unify_node_list.append(unify_node)
        
    # Create evidence job. ALWAYS HAS con_node as parent, guarantees it happens at the right time, same time as unify, NO CHILDREN.
    # It is also *backward looking*
    evidence_node = pipeline.CondorDAGNode(evidence_job)
    evidence_node.add_macro("macroiteration",it-1)  # LAST ITERATION being analyzed, not this one
    evidence_node.set_retry(opts.general_retries)
    if it > it_start:
        evidence_node.add_parent(parent_fit_node)  # run evidence after previous iteration, on previous
        dag.add_node(evidence_node)

    n_group_here = n_group
    if it ==0 and n_group_first:
        n_group_here=n_group_first
        
    # Create one node per job
    n_jobs_this_time = opts.n_samples_per_job
    if it ==it_start:
        n_jobs_this_time = n_initial
    indx_max = int((1.0*n_jobs_this_time)/n_group_here)
    if indx_max*n_jobs_this_time < n_group_here:
        indx_max+=1
    if not(it==it_start and opts.first_iteration_jumpstart):   # if on first iteration ,don't do this for jumpstart
        if opts.ile_group_subdag:
            # Write the subdag
            my_ile_subdag_name = subdag_dir + "/subdag_ILE_{}".format(it)
            my_ile_subdag = pipeline.CondorDAG(log=os.getcwd())
            add_batch_ILE_nodes_to_dag(my_ile_subdag, ile_job, None, None, indx_max, n_group_here, it, n_retries=opts.ile_retries,it_start=-1)
            my_ile_subdag.set_dag_file(my_ile_subdag_name)
            my_ile_subdag.write_concrete_dag()
            my_ile_subdag_name += ".dag"
            print(" ILE subdag plan ", my_ile_subdag_name)
            # Create a high-level node for the subdag in this dag
            main_subdag =  pipeline.CondorDAGManJob(my_ile_subdag_name)
            main_analysis_node = main_subdag.create_node()
            main_analysis_node.set_retry(0)  # currently low-level RETRY takes care of things ... beware RETRY will re do all nodes!
            if parent_fit_node:
                main_analysis_node.add_parent(parent_fit_node)
            con_node.add_parent(main_analysis_node)
            dag.add_node(main_analysis_node)
        else:
            add_batch_ILE_nodes_to_dag(dag, ile_job, parent_fit_node, con_node, indx_max, n_group_here, it, n_retries=opts.ile_retries,it_start=it_start,convert_psd_node_list=convert_psd_node_list,node_list_dict=ile_node_list_per_iteration)
            # Option C: the wide ILE jobs of iteration `it` are SEEDED from the cal proposal
            # learned by the PREVIOUS iteration's pilot -> make them depend on calpilot_{it-1}
            # so cal_consolidated_{it-1}.npz exists (the seed barrier).  The ILE args carry
            # --calibration-proposal-breadcrumb .../cal_consolidated_$(macroiterationprev).npz;
            # a missing file (early iterations) falls back to the prior inside ILE.
            if calpilot_job and ((it-1) in calpilot_node_per_iteration):
                for _ile_node in ile_node_list_per_iteration[it]:
                    _ile_node.add_parent(calpilot_node_per_iteration[it-1])
            # Extrinsic handoff: the wide ILE jobs of iteration `it` are SEEDED from the
            # extrinsic proposal consolidated from the PREVIOUS iteration -> depend on
            # extrconsolidate_{it-1} so extr_consolidated_{it-1}.npz exists (the seed barrier).
            # The ILE args carry --extrinsic-proposal-breadcrumb .../extr_consolidated_$(macroiterationprev).npz;
            # a missing/empty file (early iterations) falls back to the cold default inside ILE.
            if extrconsolidate_job and ((it-1) in extrconsolidate_node_per_iteration):
                for _ile_node in ile_node_list_per_iteration[it]:
                    _ile_node.add_parent(extrconsolidate_node_per_iteration[it-1])
#       for event in np.arange(indx_max): #np.arange(n_jobs_this_time):
#         # Add task per ILE operation
#         ile_node = pipeline.CondorDAGNode(ile_job)
# #        ile_node.set_priority(JOB_PRIORITIES["ILE"])
#         ile_node.set_retry(opts.ile_retries)
#         ile_node.add_macro("macroevent", event*n_group_here)
#         ile_node.add_macro("macrongroup", n_group_here)
#         ile_node.add_macro("macroiteration", it)
#         if not(parent_fit_node is None):
#             ile_node.add_parent(parent_fit_node)
#         if it == it_start:
#             for node in convert_psd_node_list:  # for every PSD conversion job, make sure PSD is present before we run the first iteration!
#                 ile_node.add_parent(node)
#         con_node.add_parent(ile_node) # consolidate depends on all of the individual jobs
#         dag.add_node(ile_node)
#         ile_node_list_per_iteration[it].append(ile_node)

    if puff_args and puff_cadence:
     if it>it_start and it <= puff_max_it  and (it-1)%puff_cadence ==0:  # we made a puffball last iteration, so run it through ILE now
        print(" ILE jobs for puffball on iteration ", it)
        # The puffball ILE jobs run with the SAME args_ile as the normal wide ILE, so they read
        # the same iteration-(it-1) seed breadcrumb (--calibration/--extrinsic-proposal-breadcrumb).
        # The normal-ILE seed barriers (above) were applied via ile_node_list_per_iteration BEFORE
        # these jobs existed, so thread the same barriers here via extra_parent_nodes: the puffball
        # ILE must also wait for last_puff_node (the puff side path) AND, when pilots/handoff are
        # active, for the it-1 consolidation that PRODUCES the seed file (else the puff jobs race
        # it and silently fall back to the prior).
        _puff_extra = [last_puff_node]
        if calpilot_job and ((it-1) in calpilot_node_per_iteration):
            _puff_extra.append(calpilot_node_per_iteration[it-1])
        if extrconsolidate_job and ((it-1) in extrconsolidate_node_per_iteration):
            _puff_extra.append(extrconsolidate_node_per_iteration[it-1])
        add_batch_ILE_nodes_to_dag(dag, ilePuff_job, parent_fit_node, con_node, indx_max, n_group_here, it, n_retries=opts.ile_retries,node_list_dict=ile_node_list_per_iteration,extra_parent_nodes=_puff_extra)
        # for event in np.arange(indx_max):
        #     ile_node = pipeline.CondorDAGNode(ilePuff_job)  # only difference is here: uses puffball, which by construction is the same size/ perturbed points
        #     ile_node.set_retry(opts.ile_retries)
        #     ile_node.add_macro("macroevent", event*n_group_here)
        #     ile_node.add_macro("macrongroup", n_group_here)
        #     ile_node.add_macro("macroiteration", it)
        #     if not(parent_fit_node is None):
        #         ile_node.add_parent(parent_fit_node)
        #     con_node.add_parent(ile_node) # consolidate depends on all of the individual jobs
        #     dag.add_node(ile_node)
        #     ile_node_list_per_iteration[it].append(ile_node)
        
    if fetch_args:
      if not(it==it_start and opts.first_iteration_jumpstart):   # if on first iteration ,don't do this for jumpstart 
        print(" Fetching active for iteration ", it)  # must fetch before we can do the job
        fetch_node = pipeline.CondorDAGNode(fetch_job)
        fetch_node.set_retry(opts.ile_retries)
        fetch_node.add_macro("macroiteration", it)
        if not(parent_fit_node is None):
            fetch_node.add_parent(parent_fit_node)
        parent_fit_node=fetch_node
        dag.add_node(fetch_node)

        print(" ILE jobs for fetch on iteration ", it)
        add_batch_ILE_nodes_to_dag(dag, ileFetch_job, parent_fit_node, con_node, indx_max, n_group_here, it, n_retries=opts.ile_retries,node_list_dict=ile_node_list_per_iteration)
        # for event in np.arange(indx_max):
        #     ile_node = pipeline.CondorDAGNode(ileFetch_job)  # only difference is here: uses fetch. ASSUME it is the same size (!!) -- should make this controllable
        #     ile_node.set_retry(opts.ile_retries)
        #     ile_node.add_macro("macroevent", event*n_group_here)
        #     ile_node.add_macro("macrongroup", n_group_here)
        #     ile_node.add_macro("macroiteration", it)
        #     if not(parent_fit_node is None):
        #         ile_node.add_parent(parent_fit_node)
        #     con_node.add_parent(ile_node) # consolidate depends on all of the individual jobs
        #     dag.add_node(ile_node)
        #     ile_node_list_per_iteration[it].append(ile_node)
            
    # add con job
    dag.add_node(con_node)
    dag.add_node(unify_node)
    
    # Create fit node, which depends on consolidate node
    cip_job = cip_job_list[it]
#    print(it, cip_job,creating_subdag_arg_list)
    last_cip_subdag = False
    if isinstance(cip_job,list):
        cip_worker_job = cip_job[1]
        cip_job=cip_job[0]
    elif isinstance(cip_job, str):
        last_cip_subdag = True
        print(" ==> attempting indefinite convergence subdag : {} <== ".format(it))
        # Subdag procedure, with indefinite convergence being requested
        # Write modified CIP arg string to file!
        #   - problem, need to know which CIP arg list to use.  Assume list exists
        subdag_args_now = creating_subdag_arg_list[0]; creating_subdag_arg_list = creating_subdag_arg_list[1:]
        cip_args_here = '{} '.format(opts.n_iterations_subdag_max) + subdag_args_now
        fname_subdag_cip_args = "iteration_{}_cip/args_cip.txt".format(it)
        with open(fname_subdag_cip_args,'w') as f:
            f.write(cip_args_here)
        # Create dag inside directory, as usual,  pointing to existing run (note grid size issue problem)
        cmd = "create_event_parameter_pipeline_BasicIteration --use-full-submit-paths --first-iteration-jumpstart "  # we don't use overlap-grid-0.xml.gz !
        # note we will use subdags, and do this actively.  The 0th iteration needs a SPECIFIC grid though!
        cmd += " --ile-n-events-to-analyze {} --input-grid {} --ile-exe {} --ile-args {} ".format(opts.ile_n_events_to_analyze,opts.working_directory+"/overlap-grid-0.{}".format(grid_suffix),opts.ile_exe, opts.ile_args)
        cmd += " --cip-args {}/iteration_{}_cip/args_cip.txt ".format(opts.working_directory,it)
        if opts.convert_args:
            cmd += "  --convert-args {} ".format(opts.convert_args) # passthrough, important for tides
        cmd += " --n-samples-per-job {} ".format(opts.n_samples_per_job) 
        cmd += " --neff-threshold {} ".format(opts.neff_threshold)
        cmd += " --general-request-disk {} --ile-request-disk {} --cip-request-disk {} ".format(opts.general_request_disk,opts.ile_request_disk,opts.cip_request_disk)
        cmd += " --request-memory-ILE {} --request-memory-CIP {} ".format(opts.request_memory_ILE,opts.request_memory_CIP)
        if opts.use_eccentricity:
            cmd += " --use-eccentricity "
            if opts.use_meanPerAno:
                cmd += " --use-meanPerAno "
        if opts.cip_explode_jobs:
            cmd += " --cip-explode-jobs {} ".format(opts.cip_explode_jobs)
        if opts.cip_explode_jobs_dag:   # should be deafult
            cmd += " --cip-explode-jobs-dag "  
        if opts.cip_explode_jobs_subdag:   # should be deafult
            cmd += " --cip-explode-jobs-subdag "  
        if opts.request_gpu_ILE:
            cmd += " --request-gpu-ILE "  # DAG writer needs to make appropririate requests
        if opts.request_xpu_ILE:
            cmd += " --request-xpu-ILE "  # DAG writer needs to make appropririate requests
        if opts.use_tabular_eos_file:
            cmd += " --use-tabular-eos-file "
        if opts.ile_retries:
            cmd += " --ile-retries {} ".format(opts.ile_retries)
        if opts.general_retries:
            cmd += " --general-retries {} ".format(opts.general_retries)
        if opts.use_singularity:
            cmd += " --use-singularity "
        if opts.cache_file:
            cmd += " --cache-file {} ".format(opts.cache_file)
        if opts.frames_dir:
            cmd += " --frames-dir {} ".format(opts.frames_dir)
        if opts.ile_runtime_max_minutes:
            cmd += " --ile-runtime-max-minutes {} ".format(opts.ile_runtime_max_minutes)
        if opts.ile_condor_commands:
            cmd += " --ile-condor-commands {} ".format(opts.ile_condor_commands)
        if opts.use_osg:
            cmd += " --use-osg "
            if opts.use_osg_cip:
                cmd += " --use-osg-cip "
            if opts.use_cvmfs_frames:
                cmd += " --use-cvmfs-frames "
            if opts.use_osg_simple_requirements:
                cmd +=  " --use-osg-simple-requirements "
            if opts.condor_local_nonworker:
                cmd += " --condor-local-nonworker "
            if opts.condor_local_nonworker_igwn_prefix:
                cmd += " --condor-local-nonworker-igwn-prefix "
            if opts.condor_nogrid_nonworker:
                cmd += " --condor-nogrid-nonworker "
            if opts.transfer_file_list:
                cmd += " --transfer-file-list {} ".format(opts.transfer_file_list)
        if opts.test_args:
            import re
            test_args_optin = "X " + test_args  # copy not reference
            test_args_optin = re.sub('--always-succeed', '', test_args_optin)  # tests must operate
            # remove iteration threshold, so they ALWAYS operate
            test_args_optin = re.sub('--iteration-threshold [1-9] ', '', test_args_optin)
        else:
            test_args_optin = "X  --parameter m1 --method lame "
        if opts.puff_args and opts.puff_cadence:
            # always puff in the final stage, it can be dangerous not to.  ALL iterations
            cmd += " --puff-exe {} --puff-args {} --puff-cadence {} --puff-max-it {}  ".format(opts.puff_exe, opts.puff_args, opts.puff_cadence,opts.n_iterations_subdag_max) 
        fname_subdag_test_args = opts.working_directory+"/iteration_{}_cip/args_test.txt".format(it)
        with open(fname_subdag_test_args,'w') as f:
            f.write(test_args_optin)
        cmd+= " --test-args {}/iteration_{}_cip/args_test.txt ".format(opts.working_directory,it)
        cmd += " --working-directory {}/iteration_{}_cip/ ".format(opts.working_directory,it)
        cmd += " --n-iterations {} ".format(opts.n_iterations_subdag_max) # default will be 10 iterations, but argument controls
        # 
        print(" SUBDAG COMMAND : ", cmd)
        os.system(cmd)
        # link the *input* grid in, overwriting the (temporary, placeholder) grid used to create the workflow
        cmd = "ln -sf {}/overlap-grid-{}.{suf} iteration_{}_cip/overlap-grid-0.{suf}".format(opts.working_directory,it,it, suf=grid_suffix)
        os.system(cmd)

        main_subdag =  pipeline.CondorDAGManJob("{}/iteration_{}_cip/marginalize_intrinsic_parameters_BasicIterationWorkflow.dag".format(opts.working_directory,it))  # assumes subdag is created

        # Create viable job
        main_analysis_node = main_subdag.create_node()
        main_analysis_node.add_parent(parent_fit_node)
        main_analysis_node.set_retry(opts.general_retries)

        # Add post script OR NODE to fetch information from the job
        #   - problem: dagman jobs don't allow post scripts usually, may need to write this manually
        fname_fetch = opts.working_directory+"/"+"internal_fetch_from_{}.json".format(it)
        my_dict = {"method":"native", "source":opts.working_directory+"/iteration_{}_cip".format(it),"n_max":3000}
        with open(fname_fetch,'w') as f:
            json.dump(my_dict,f)
        fetch_subdag_args =" --inj-file-out overlap-grid-{}.{} ".format(it+1, grid_suffix)
        fetch_subdag_args += " --input-json {} ".format(fname_fetch)
        fetch_subdag_job, fetch_subdag_job_name = dag_utils.write_puff_sub(tag='FETCH_{}_subdag'.format(it),log_dir=opts.working_directory+"/"+"iteration_{}_cip/logs/".format(it),arg_str=fetch_subdag_args,request_memory=opts.request_memory_ILE,input_net=None,output=opts.working_directory+'/overlap-grid-$(macroiterationnext).{}'.format(grid_suffix),out_dir=opts.working_directory,exe=dag_utils.which("util_FetchExternalGrid.py"),universe=local_worker_universe,no_grid=no_worker_grid)
        fetch_subdag_job.add_condor_cmd('request_disk',opts.general_request_disk)
        fetch_subdag_job.write_sub_file()

        # Because later we are going to take care of the iteration crap and parents for the 
        dag.add_node(main_analysis_node)
        cip_node= main_analysis_node
        cip_node.add_parent(unify_node)
        # increment definitions, so the code later works fine
        unify_node=main_analysis_node
        cip_job = fetch_subdag_job  # we will create this as a job and make its parent equal to above

        # Create link to HARVEST ALL COMPOSITE DATA generated during subdag operation, so we don't need to regenerate it
        cmd_fetch_composite = "ln -sf {}/iteration_{}_cip/all.net {}/bonus_subdag_{}.composite ".format(opts.working_directory, it,opts.working_directory,it)
        os.system(cmd_fetch_composite)


        # Create link to FEED  COMPOSITE DATA generated during top-level run into low-level directory
        # Make sure we are only linking *prior* iterations information, for clarity of provenance
        for indx in np.arange(it+1): # we still do one ILE iteration beforehand at top level, rather than doing it in subdag, because CIP follows ILE
            cmd_fetch_composite = "ln -sf {}/consolidated_{}.composite {}/iteration_{}_cip/input_{}.composite   ".format(opts.working_directory, indx,  opts.working_directory, it,indx)
            os.system(cmd_fetch_composite)


        print(" ==> end setup convergence subdag : {} <== ".format(it))
        # Need postprocessing, based on previous stage

        # if isinstance(cip_job_list[it-1],list):
        #     cip_job = cip_job_list[it-1][0]
        # else:
        #     cip_job= cip_job_list[it-1]

    # Node generation for cip job tasks
    if not(isinstance(cip_job,str)):
        fit_node = pipeline.CondorDAGNode(cip_job)
    else:
        fit_node = main_analysis_node
    fit_node.add_macro("macroiteration", it)
    fit_node.add_macro("macroiterationnext", it+1)
    fit_node.set_category("CIP")
    fit_node.add_parent(unify_node)  # only fit if we have results from the previous iteration
    fit_node.set_retry(opts.general_retries)
    # skip CIP node creation if we have worker nodes! Note this will BREAK GP, so also use it for GP
    if (opts.cip_explode_jobs is None) or (last_cip_subdag) or ('fit-method gp' in cip_args_base):
        dag.add_node(fit_node) 
        parent_fit_node = fit_node
    else:
        parent_fit_node = unify_node
    if not(opts.cip_explode_jobs is None) and not(last_cip_subdag): # don't explode if we just made a subdag!
        print(" Exploding workers out ")
        # Create job to consolidate worker outputs
        join_node =pipeline.CondorDAGNode(join_cip_job)
        join_node.add_macro("macroiteration", it)
        join_node.add_macro("macroiterationnext", it+1)
        join_node.set_category("join_cip")
        join_node.set_retry(opts.general_retries)
        
        # Create exploded worker job nodes
        if opts.cip_explode_jobs is None:
            worker_node =pipeline.CondorDAGNode(cip_worker_job)
            worker_node.add_macro("macroiteration", it)
            worker_node.add_macro("macroiterationnext", it+1)
            worker_node.set_category("CIP_worker")
            worker_node.add_parent(parent_fit_node)  # only fit if we have results from the previous iteration
            worker_node.set_retry(opts.general_retries)
            join_node.add_parent(worker_node)  # make sure to add worker node as parent
            dag.add_node(worker_node)
        elif opts.cip_explode_jobs_subdag:
            n_explode= int(opts.cip_explode_jobs)
            # write appropriate subdag
            subdag_name = dag_utils.write_CIP_single_iteration_subdag(cip_worker_job, it, str(it), subdag_dir,n_explode=n_explode)
            print(" Subdag plan ", subdag_name)
            # Create node
            main_subdag =  pipeline.CondorDAGManJob(subdag_name)
            main_analysis_node = main_subdag.create_node()
            main_analysis_node.set_category("cip_subdag")
            main_analysis_node.set_retry(1000) # some enormous number, should be user specified
            # add links
            main_analysis_node.add_parent(parent_fit_node)
            join_node.add_parent(main_analysis_node)
            # register subdag, so we can add SCRIPT POST at the end
            cip_subdag_nodes[it] = main_analysis_node
            cip_subdag_work_targets[it] = opts.n_samples_per_job
            # add to DAG
            dag.add_node(main_analysis_node)
        else:
            n_explode = opts.cip_explode_jobs
            # if we are on the last iteration and we want to use more exploded jobs now, explode more jobs
            if opts.cip_explode_jobs_last and (it == opts.n_iterations-1): # last iteration
                print("   Last iteration explode size ", opts.cip_explode_jobs_last)
                n_explode = opts.cip_explode_jobs_last
            for indx in np.arange(n_explode):
                worker_node =pipeline.CondorDAGNode(cip_worker_job)
                worker_node.add_macro("macroiteration", it)
                worker_node.add_macro("macroiterationnext", it+1)
                worker_node.set_category("CIP_worker")
                worker_node.add_parent(parent_fit_node)  # only fit if we have results from the previous iteration
                worker_node.set_retry(opts.general_retries)
                join_node.add_parent(worker_node)  # make sure to add worker node as parent
                dag.add_node(worker_node)
        dag.add_node(join_node)
        parent_fit_node=join_node

    # Check if puffball being created, and if so create it *immediately* after CIP job creates the overlap-grid file
    if puff_args and puff_cadence:
      if it > -1 and it <= puff_max_it and it%puff_cadence ==0:
        print(" Puffball for iteration ", it)
        puff_node = pipeline.CondorDAGNode(puff_job)
        puff_node.add_macro("macroiteration", it)
        puff_node.add_macro("macroiterationnext", it+1)
        puff_node.set_category("PUFF")
        puff_node.set_retry(opts.general_retries)
        if not (parent_fit_node is None):
            puff_node.add_parent(parent_fit_node)  # only fit if we have results from the previous iteration
        dag.add_node(puff_node)
        # PUFF FIX (c4c1455e): the puffball node is a barrier for the NEXT iteration's PUFFBALL
        # ILE jobs ONLY -- it must NOT be folded into parent_fit_node (the normal fit chain), or
        # the next iteration's NORMAL ILE jobs would wait on the puff side path and the pipeline
        # halts at the PUFF stage.  Keep parent_fit_node = the fit chain; carry the puff barrier
        # separately in last_puff_node, applied to ilePuff jobs via extra_parent_nodes.
        last_puff_node = puff_node

    # LISA reflected-sky-mode: at the chosen iteration, reflect the grid CIP just
    # produced (overlap-grid-(it+1)) BEFORE the next iteration's ILE reads it.
    # Unlike puffball this MUST gate the next ILE (it rewrites the grid in place),
    # so fold it into parent_fit_node.
    if (reflected_mode_job is not None) and (it == reflected_mode_iter):
        print(" Reflected-sky-mode for iteration ", it)
        reflect_node = pipeline.CondorDAGNode(reflected_mode_job)
        reflect_node.add_macro("macroiteration", it)
        reflect_node.add_macro("macroiterationnext", it+1)
        reflect_node.set_category("CONVERT")
        reflect_node.set_retry(opts.general_retries)
        if not (parent_fit_node is None):
            reflect_node.add_parent(parent_fit_node)
        dag.add_node(reflect_node)
        parent_fit_node = reflect_node

    # Calibration PILOT for this iteration (Option C).  Runs IN PARALLEL with CIP/puff --
    # it does NOT gate them (parent_fit_node is left untouched) -- harvesting iteration
    # it's composite (via unify_node, which guarantees a non-empty composite) and emitting
    # cal_consolidated_<it>.npz, which SEEDS the wide ILE jobs of iteration it+1.
    if calpilot_job and (it <= opts.calmarg_pilot_max_it) and (it % opts.calmarg_pilot_cadence == 0):
        print(" Calibration pilot for iteration ", it)
        calpilot_node = pipeline.CondorDAGNode(calpilot_job)
        calpilot_node.add_macro("macroiteration", it)
        calpilot_node.add_macro("macroiterationprev", it-1)
        calpilot_node.set_category("CALPILOT")
        calpilot_node.set_retry(opts.general_retries)
        calpilot_node.add_parent(unify_node)   # need iteration it's (non-empty) composite
        dag.add_node(calpilot_node)
        calpilot_node_per_iteration[it] = calpilot_node

    # Extrinsic handoff consolidation for this iteration.  Runs IN PARALLEL with CIP/puff
    # (does NOT gate them) -- it depends on unify_node, which guarantees all of iteration
    # it's ILE jobs finished (so the per-event extr_proposal_<it>_*.npz are present), and
    # emits extr_consolidated_<it>.npz, which SEEDS the wide ILE jobs of iteration it+1.
    if extrconsolidate_job:
        extrconsolidate_node = pipeline.CondorDAGNode(extrconsolidate_job)
        extrconsolidate_node.add_macro("macroiteration", it)
        extrconsolidate_node.set_category("EXTRCONSOLIDATE")
        extrconsolidate_node.set_retry(opts.general_retries)
        extrconsolidate_node.add_parent(unify_node)   # all of iteration it's ILE jobs done
        dag.add_node(extrconsolidate_node)
        extrconsolidate_node_per_iteration[it] = extrconsolidate_node


    if _use_hpip_pipeline:
        alias_grid_node = pipeline.CondorDAGNode(alias_grid_job)
        alias_grid_node.add_macro("macroiteration", it+1)
        alias_grid_node.add_parent(parent_fit_node)
        alias_grid_node.set_category("ALIAS")
        alias_grid_node.set_retry(opts.general_retries)
        dag.add_node(alias_grid_node)
        parent_fit_node = alias_grid_node

    # Create convert node, which depends on fit node, *if* tests are being performed.
    # In hyperpipeline mode an alias node creates posterior_samples-*.dat
    # symlinks to the header-bearing overlap-grid-*.dat files instead.
    if opts.test_args and not _use_hpip_pipeline:
        convert_node=pipeline.CondorDAGNode(convert_job)
        convert_node.add_macro("macroiteration", it+1)  # convert the NEWLY-PRODUCED iteration
        convert_node.add_macro("macroiterationlast", it)  # use log files in the previous directory
        convert_node.add_parent(parent_fit_node)
        convert_node.set_category("CONVERT")
        convert_node.set_retry(opts.general_retries)
        dag.add_node(convert_node)

        parent_fit_node = convert_node



    if opts.test_args and it>0:
        # Cannot run test on first iteration
        test_node = pipeline.CondorDAGNode(test_job)
        test_node.add_macro("macroiteration", it+1)   # test the NEWLY-PRODUCED iteration against the old
        test_node.add_macro("macroiterationlast", it)
        test_node.add_parent(parent_fit_node)
        test_node.set_category("CONVERGE")
        dag.add_node(test_node)
        test_node_list.append(test_node)

        parent_fit_node=test_node


# Create export stages for extrinsic samples
last_node= parent_fit_node # backstop
if opts.last_iteration_extrinsic:
    # Check if 'it' is defined : it will not always be, if done later
    if not ('it' in globals()):
        it = opts.n_iterations  # last iteration
    elif it < opts.n_iterations and 'Z' in cip_args_prefixes:
        it += 1   # deal with an offset/fencepost issue that can happen when using subdags for some reason
    print(" Extrinsic information being pulled from iteration {} ".format(it))

    # Create nodes for followup tasks
    cat_node = pipeline.CondorDAGNode(cat_job)
    cat_node.add_macro("macroiteration", it)  # needed to identify log file location

    # Per-distance likelihood export consolidation nodes (depend directly on
    # ILE_extr, since .dgrid / .dslice are emitted by ILE itself with no
    # convert/resample step in between).
    dgrid_node = dslice_node = None
    if opts.last_iteration_export_marginal_distance_grid:
        dgrid_node = pipeline.CondorDAGNode(dgrid_job)
        dgrid_node.add_macro("macroiteration", it)
        dgrid_node.set_retry(opts.ile_retries)
    if opts.last_iteration_export_distance_slices and opts.last_iteration_export_distance_slices > 0:
        dslice_node = pipeline.CondorDAGNode(dslice_job)
        dslice_node.add_macro("macroiteration", it)
        dslice_node.set_retry(opts.ile_retries)

    # Perform final ILE run on all points, saving samples
    # Need to perform number of events CONSISTENT WITH TARGET SAMPLE SIZE
    #    - *not* always same as number of ILE events being analyzed
    #    - *assumes* grid files have sufficiently large numbers of samples to allow this! (as in many other cases)
    n_jobs_extrinsic = int(opts.last_iteration_extrinsic_nsamples/(1.0*n_group))

    if opts.last_iteration_extrinsic_time_resampling:
        convert_node = pipeline.CondorDAGNode(batchConvertExtr_job)
        convert_node.set_retry(opts.ile_retries)  # this can fail too
        convert_node.add_macro("macroiteration",it)  # needed so we find the correct data to read


    for event in np.arange(n_jobs_extrinsic):
        # Add task per ILE operation
        ile_node = pipeline.CondorDAGNode(ileExtr_job)
#        ile_node.set_priority(JOB_PRIORITIES["ILE"])
        ile_node.set_retry(opts.ile_retries)
        ile_node.add_macro("macroevent", event*n_group)
        ile_node.add_macro("macrongroup", n_group)
        ile_node.add_macro("macroiteration", it)
        if not(parent_fit_node is None):
            ile_node.add_parent(parent_fit_node)
        dag.add_node(ile_node)
#        ile_node_list_per_iteration[it].append(ile_node)

        # Distance-grid / -slice consolidation runs after EVERY ILE_extr
        # finishes, since each ILE_extr emits one .dgrid / .dslice file.
        if dgrid_node is not None:
            dgrid_node.add_parent(ile_node)
        if dslice_node is not None:
            dslice_node.add_parent(ile_node)

        # Add convert and resample task *for each output file*
        if opts.last_iteration_extrinsic_time_resampling:
            # establish parent-child relationship
            convert_node.add_parent(ile_node)
        elif opts.last_iteration_extrinsic_batched_convert:
            convert_node = pipeline.CondorDAGNode(batchConvertExtr_job)
            convert_node.add_macro("macroevent", event*n_group)
            convert_node.add_macro("macrongroup", n_group)
            convert_node.add_macro("macroiteration", it)
            convert_node.set_retry(opts.ile_retries)  # this can fail too
            convert_node.add_parent(ile_node)
            dag.add_node(convert_node)
            cat_node.add_parent(convert_node)
        else:
          for indx in np.arange(n_group):
            convert_node = pipeline.CondorDAGNode(convertExtr_job)
            convert_node.add_macro("macroevent", event*n_group)
            convert_node.add_macro("macroiteration", it)
            convert_node.add_macro("macroindx",indx)
            convert_node.set_retry(opts.ile_retries)  # this can fail too
            convert_node.add_parent(ile_node)

            resample_node = pipeline.CondorDAGNode(resample_job)
            resample_node.add_macro("macroevent", event*n_group)
            resample_node.add_macro("macroiteration", it)
            resample_node.add_macro("macroindx",indx)
            resample_node.set_retry(opts.ile_retries)  # these occasionally fail for stupid reasons - nodes missing software, etc
            resample_node.add_parent(convert_node)
            
            # Make cat job 
            cat_node.add_parent(resample_node)
            cat_node.set_retry(opts.ile_retries)  # this can fail too

            # Add nodes
            dag.add_node(convert_node)
            dag.add_node(resample_node)
   
    if not(opts.last_iteration_extrinsic_time_resampling):
        dag.add_node(cat_node)
        last_node=cat_node
    else:
        dag.add_node(convert_node)   # this is the time resampling task
        last_node=convert_node
    if dgrid_node is not None:
        dag.add_node(dgrid_node)
    if dslice_node is not None:
        dag.add_node(dslice_node)

# Create rotation job (if extrinsic available)
if opts.frame_rotation and opts.last_iteration_extrinsic:
    rotate_node = pipeline.CondorDAGNode(rotate_job)
    rotate_node.set_category("ROTATE")
    rotate_node.add_parent(last_node)
    last_node=rotate_node
    dag.add_node(rotate_node)

# Creating calibration uncertainty job
if opts.calibration_reweighting:
    if opts.bilby_ini_file:
        pickle_node = pipeline.CondorDAGNode(pickle_job)
        pickle_node.add_macro("macroiteration",it)
        pickle_node.set_category("PICKLE")
        pickle_node.add_parent(last_node)
        last_node=pickle_node
        dag.add_node(pickle_node)
    # Do single job if only one needed, otherwise loop
    if not(opts.calibration_reweighting_batchsize):
        calibration_node = pipeline.CondorDAGNode(calibration_job)
        calibration_node.add_macro("macroiteration",it)
        calibration_node.set_category("CALIBRATION")
        calibration_node.add_parent(last_node)
        calibration_node.set_retry(opts.general_retries)
        dag.add_node(calibration_node)
    else:
        # Make job to reweight everything, after producing samples
        reweight_merge_node =    pipeline.CondorDAGNode(reweight_merge_job)
        reweight_merge_node.add_macro("macroiteration",it)

        # Make individual calibration nodes
        for start_indx in np.arange(start=0,stop=opts.last_iteration_extrinsic_nsamples,step=opts.calibration_reweighting_batchsize):
            end_indx = start_indx + opts.calibration_reweighting_batchsize  # end index is not included, python-style indexing so no fencepost problems
            calibration_node = pipeline.CondorDAGNode(calibration_job)
            calibration_node.add_macro("macroiteration",it)
            calibration_node.add_macro("macrostartidx",start_indx)
            calibration_node.add_macro("macroendidx",end_indx)
            calibration_node.set_category("CALIBRATION")
            calibration_node.set_retry(opts.general_retries)
            calibration_node.add_parent(last_node)
            reweight_merge_node.add_parent(calibration_node)
            dag.add_node(calibration_node)
            
        # add the merge node.  Thouls be the last node
        dag.add_node(reweight_merge_node)
    
if opts.comov_distance_reweighting:
    convert_ascii_node = pipeline.CondorDAGNode(convert_ascii_job)
    convert_ascii_node.add_macro("macroiteration",it)
    convert_ascii_node.set_category("CONVERT_ASCII")
    convert_ascii_node.add_parent(last_node)
    dag.add_node(convert_ascii_node)
    distance_node = pipeline.CondorDAGNode(distance_job)
    distance_node.add_macro("macroiteration",it)
    distance_node.set_category("COMOV")
    distance_node.add_parent(convert_ascii_node)
    last_node=distance_node
    dag.add_node(distance_node)
# Create final node for overall plots.  (Note: default setup is designed to enable plots of the last two iterations *at each step* but this seems like overkill)
if plot_args:
        # Cannot run test on first iteration
        plot_node = pipeline.CondorDAGNode(plot_job)
        plot_node.add_macro("macroiteration", it)
        plot_node.add_macro("macroiterationlast", it-1)
        plot_node.add_parent(parent_fit_node)
        plot_node.set_category("PLOT")
        dag.add_node(plot_node)

dag_name="marginalize_intrinsic_parameters_BasicIterationWorkflow"
dag.set_dag_file(dag_name)
dag.write_concrete_dag()

# Add control-logic overlays through the workflow abstraction.  The
# active backend (htcondor / glue / slurm) translates these into its
# native form: HTCondor and glue.pipeline emit DAGMan SCRIPT POST /
# ABORT-DAG-ON / DOT directives unchanged; Slurm emits afterany-
# dependency wrapper sbatch jobs and a graphviz file.
# https://htcondor.readthedocs.io/en/latest/users-manual/dagman-workflows.html
if opts.test_args:
    for node in test_node_list:
        dag.add_abort_on(node, exit_code=1, return_value=0)

indx = 0  # start with consolidated_0
unify_nodes_to_post = list(unify_node_list)
if opts.first_iteration_jumpstart:
    unify_nodes_to_post = unify_nodes_to_post[1:]  # drop first element
    indx = 1
with open("confirm_exist.sh",'w') as f:
    f.write("""#! /bin/bash
[ -s $1 ]
""")
st=os.stat("confirm_exist.sh"); os.chmod("confirm_exist.sh", st.st_mode | stat.S_IEXEC)
for indx_here, node in enumerate(con_node_list):
    if indx_here >= indx:  # skip 0th entry if appropriate
        dag.add_script_post(
            node,
            "{}/confirm_exist.sh".format(opts.working_directory),
            "{}/consolidated_{}.composite".format(opts.working_directory, indx_here),
        )
for node in unify_nodes_to_post:
    dag.add_script_post(
        node,
        "{}/confirm_exist.sh".format(opts.working_directory),
        "{}/all.net".format(opts.working_directory),
    )
# add dot file, for dag visualization
dag.set_dot_file('my_dag_vis.dot')


# CIP adaptive batch size per iteration.  BACKSTOP with 'default' script to check work by lilne-counting (dangerous!)
if opts.cip_explode_jobs_subdag:
    if opts.cip_post_exe is None:
    # script to count lines in overlap-grid files. Note this does NOT check n_eff directly ... should count n_eff instead based on +annotation.dat files
        with open("cip_check_work.sh", 'w') as f:
            if _use_hpip_pipeline:
                # Hyperpipeline: count non-comment data rows across all
                # overlap-grid-*.dat shards in the directory.  No
                # igwn_ligolw_print dependency.
                f.write("""#! /bin/bash
LINES=`grep -hE -v '^#' $1/overlap-grid-*.dat 2>/dev/null | grep -cv '^[[:space:]]*$'`
[ ${LINES} -ge $2 ]
""")
            else:
                f.write("""#! /bin/bash
LINES=`igwn_ligolw_print -c mass1 $1/overlap-grid*.xml.gz | wc -l `
[ ${LINES} -ge $2 ]
""")
        st=os.stat("cip_check_work.sh"); os.chmod("cip_check_work.sh", st.st_mode | stat.S_IEXEC)
        opts.cip_post_exe = opts.working_directory + "/cip_check_work.sh"
    # add POST SCRIPT to subdag. Note it will USUALLY FAIL and get retried A LARGE NUMBER OF TIMES, until we complete work!
    for it in cip_subdag_nodes:
        node = cip_subdag_nodes[it]
        dag.add_script_post(
            node, opts.cip_post_exe,
            "iteration_{}_cip".format(it),
            cip_subdag_work_targets[it],
        )

# ILE adaptive batch size per iteration: If ile_post_exe exists, write the necessary items
# WARNING: currently framed as a post script that is run after EVERY job
if opts.ile_post_exe:
    for it in ile_node_list_per_iteration:
        for node in ile_node_list_per_iteration[it]:
            # write the script post line. Add in the worklevel target
            n_target = n_initial
            if opts.n_samples_per_job_threshold:
                n_target = opts.n_samples_per_job_threshold
            dag.add_script_post(
                node, opts.ile_post_exe,
                "$(JOBID)", "$(RETURN)", str(it), str(n_target),
            )
            
        
