From 10cab2f3340575763a2c8f3dc876ac000c00085d Mon Sep 17 00:00:00 2001
From: Surabhi Sachdev <surabhi.sachdev@comp-hd-002.gwave.ics.psu.edu>
Date: Mon, 26 Oct 2020 11:53:33 -0400
Subject: [PATCH] extractor_dag_patch

---
 gstlal-inspiral/bin/gstlal_inspiral                |   4 +-
 .../bin/gstlal_inspiral_coinc_extractor_dag        | 114 +++++++++++++++------
 gstlal-inspiral/python/inspiral.py                 |   2 +-
 gstlal-inspiral/python/lloidhandler.py             |  14 ++-
 gstlal-inspiral/python/streamthinca.py             |   4 +-
 5 files changed, 95 insertions(+), 43 deletions(-)

diff --git a/gstlal-inspiral/bin/gstlal_inspiral b/gstlal-inspiral/bin/gstlal_inspiral
index b63f232fb..da588b12a 100755
--- a/gstlal-inspiral/bin/gstlal_inspiral
+++ b/gstlal-inspiral/bin/gstlal_inspiral
@@ -294,6 +294,7 @@ def parse_command_line():
 	group.add_option("--local-frame-caching", action = "store_true", help = "Pre-reads frame data, performs downsampling, and stores to local filespace. ")
 	group.add_option("--nxydump-segment", metavar = "start:stop", default = ":", help = "Set the time interval to dump from nxydump elments (optional).  The default is \":\", i.e. dump all time.")
 	group.add_option("--reconstruction-segment", metavar = "start:stop", action = "append", help = "Only reconstruct the SNRs for this time interval (optional). Can be provided multiple times.")
+	group.add_option("--rank-template-id", metavar = "int", type = "int", help = "Only assign ranking statistic values to candidates associated with this template id. This only affects runs where candidates have ranking statistic values assigned on-the-fly.")
 	group.add_option("-t", "--tmp-space", metavar = "path", help = "Path to a directory suitable for use as a work area while manipulating the database file.  The database file will be worked on in this directory, and then moved to the final location when complete.  This option is intended to improve performance when running in a networked environment, where there might be a local disk with higher bandwidth than is available to the filesystem on which the final output will reside.")
 	group.add_option("-v", "--verbose", action = "store_true", help = "Be verbose (optional).")
 	group.add_option("--write-pipeline", metavar = "filename", help = "Write a DOT graph description of the as-built pipeline to this file (optional).  The environment variable GST_DEBUG_DUMP_DOT_DIR must be set for this option to work.")
@@ -825,7 +826,7 @@ for output_file_number, (svd_bank_url_dict, output_url, ranking_stat_output_url,
 				rankingstat.template_ids = template_ids
 			elif rankingstat.template_ids != template_ids:
 				raise ValueError("\"%s\" is for the wrong templates" % options.ranking_stat_input)
-			rankingstat.numerator.set_horizon_factors(horizon_factors)
+#			rankingstat.numerator.set_horizon_factors(horizon_factors)
 
 			if options.compress_ranking_stat:
 				if options.verbose:
@@ -891,6 +892,7 @@ for output_file_number, (svd_bank_url_dict, output_url, ranking_stat_output_url,
 		cluster = True,#options.data_source in ("lvshm", "framexmit"),# If uncommented, we only cluster when running online
 		cap_singles = options.cap_singles,
 		FAR_trialsfactor = options.far_trials_factor,
+		template_id = options.rank_template_id,
 		activation_counts = options.activation_counts,
 		template_id_time_map = options.upload_time_before_merger,
 		verbose = options.verbose
diff --git a/gstlal-inspiral/bin/gstlal_inspiral_coinc_extractor_dag b/gstlal-inspiral/bin/gstlal_inspiral_coinc_extractor_dag
index 695e1bbac..40add1375 100755
--- a/gstlal-inspiral/bin/gstlal_inspiral_coinc_extractor_dag
+++ b/gstlal-inspiral/bin/gstlal_inspiral_coinc_extractor_dag
@@ -19,9 +19,12 @@ def parse_command_line():
 	parser = OptionParser()
 	parser.add_option("-i", "--injection-database", metavar = "filename", help = "Path to the injection database.")
 	parser.add_option("-z", "--zerolag-database", metavar = "filename", help = "Path to the zerolag database.")
+	parser.add_option("-r", "--ranking-stat-dir", metavar = "path", help = "Path to the directory containing the ranking stat files. If not provided, ranking stat files will be assumed to be in the same directory as the database provided.")
+	parser.add_option("--general-dependencies-directory", metavar = "path", help = "Path to the directory containing the reference psd, frame cache, time slide file, vetoes, segments, and svd. If not specified, assumed to be the directory the database is in.")
+	parser.add_option("--mass-model-file", metavar = "filename", help = "Path to the mass model file if it is not in gstlal_inspiral_mass_model in the directory the database is stored.")
 	parser.add_option("-t", "--tmp-space", metavar = "path", help = "Path to temp space. Defaults to _CONDOR_SCRATCH_DIR .")
 	parser.add_option("-f", "--far-threshold", type = "float", default = 3.86e-7, help = "Threshold for rerunning an injection job to produce the time series.")
-	parser.add_option("-m", "--max-number-inspiral-jobs", type = "int", default = 10000, help = "Maximum number of inspiral jobs to queue. If this option is not provided, the program defaults to the top 10000 that pass the imposed FAR threshold.")
+	parser.add_option("-m", "--max-number-inspiral-jobs", type = "int", default = 1000000, help = "Maximum number of inspiral jobs to queue. If this option is not provided, the program defaults to the top 10000 that pass the imposed FAR threshold.")
 
 	options, filenames = parser.parse_args()
 
@@ -142,6 +145,7 @@ CREATE TEMPORARY TABLE
 	AS
 		SELECT
 			sngl_inspiral.process_id AS pid,
+			sngl_inspiral.Gamma0 AS Gamma0,
 			sngl_inspiral.Gamma1 AS Gamma1,
 			sim_id_combined_far.far AS far,
 			sim_inspiral.simulation_id AS simulation_id,
@@ -169,6 +173,7 @@ CREATE TEMPORARY TABLE
 AS
 	SELECT
 		coinc_event.process_id, 
+		(SELECT sngl_inspiral.Gamma0 FROM sngl_inspiral WHERE map.event_id == sngl_inspiral.event_id LIMIT 1),
 		(SELECT sngl_inspiral.Gamma1 FROM sngl_inspiral WHERE map.event_id == sngl_inspiral.event_id LIMIT 1), 
 		combined_far AS far, 
 		coinc_inspiral.coinc_event_id,
@@ -213,11 +218,12 @@ if options.injection_database:
 	LIMIT ? 
 	""" % (options.far_threshold,), (int(num_inspiral_jobs),)):
 		process_id = record[0]
-		bank_id = record[1]
-		far = record[2]
-		sid = record[3]
-		srow = record[4:]
-		s_row[(bank_id, process_id, sid)] = inspiral_table.row_from_cols(srow)
+		template_id = record[1]
+		bank_id = record[2]
+		far = record[3]
+		sid = record[4]
+		srow = record[5:]
+		s_row[(template_id, bank_id, process_id, sid)] = inspiral_table.row_from_cols(srow)
 
 elif options.zerolag_database:
         create_coinc_view(connection, options)
@@ -236,11 +242,12 @@ elif options.zerolag_database:
         LIMIT ? 
         """ % (options.far_threshold,), (int(num_inspiral_jobs),)):
                 process_id = record[0]
-                bank_id = record[1]
-                far = record[2]
-                sid = record[3]
-                srow = record[4:]
-                s_row[(bank_id, process_id, sid)] = coinc_inspiral_table.row_from_cols(srow)
+		template_id = record[1]
+                bank_id = record[2]
+                far = record[3]
+                sid = record[4]
+                srow = record[5:]
+                s_row[(template_id, bank_id, process_id, sid)] = coinc_inspiral_table.row_from_cols(srow)
 
 master_opts_dict = { 
 	"gps-start-time":None,
@@ -255,7 +262,7 @@ master_opts_dict = {
 	"veto-segments-name": "vetoes",
 	"fir-stride": 0.25,
 	"gracedb-group": "CBC",
-	"coincidence-threshold": 0.005,
+	"coincidence-threshold": None,
 	"control-peak-time": 0,
 	"gracedb-pipeline": "gstlal",
 	"data-source": None,
@@ -265,6 +272,8 @@ master_opts_dict = {
 	"channel-name": None,
 	"singles-threshold": "inf",
 	"local-frame-caching":"",
+	"reconstruction-segment": None,
+	"rank-template-id": None,
 	"verbose": ""
 }
 
@@ -281,8 +290,8 @@ master_input_dict = {
 	}
 
 master_output_dict = {
-	"ranking-stat-output": "not_used.xml.gz",
-	"zerolag-rankingstat-pdf": "notused2.xml.gz",
+	"ranking-stat-output": None,
+	"zerolag-rankingstat-pdf": None,
 	"output": None,
 }
 
@@ -320,9 +329,9 @@ def updatedict(x, y):
 			except KeyError as e:
 				pass
 
-def fixrelpath(x, ys):
+def fixrelpath(x, ys, d):
 	for y in ys:
-		x[y] = "%s/%s" % (analysis_dir, x[y][0])
+		x[y] = "%s/%s" % (d, x[y][0])
 
 def new_inj_file(row, output):
 	xmldoc = ligolw.Document()
@@ -358,6 +367,18 @@ try:
 except OSError:
 	pass
 
+
+if not options.mass_model_file:
+	mass_model = analysis_dir + '/gstlal_inspiral_mass_model'
+	try:
+		os.mkdir("gstlal_inspiral_mass_model")
+		os.system('cp -r %s .' % (mass_model,))
+	except OSError:
+		pass
+else:
+	mass_model = options.mass_model_file
+	os.system('cp %s .' % (mass_model,))
+
 bayesdir = "bayestar_input"
 try:
 	os.mkdir(bayesdir)
@@ -367,17 +388,21 @@ except OSError:
 
 f = open("process.sh", "w")
 f.write("""#!/usr/bin/bash
-ID=$1
-FILE=$(gstlal_inspiral_best_coinc_file lloid_files/${ID}/*CBC_AllSky-*-0.xml)
-ligolw_no_ilwdchar ${FILE}
-mkdir -p bayestar_input/${ID}/
-gstlal_ligolw_add_without_reassign ${FILE} inj_files/${ID}_inj.xml.gz --output bayestar_input/${ID}/${ID}_event.xml.gz
-lalapps_inspinjfind --time-window 0.9 bayestar_input/${ID}/${ID}_event.xml.gz
-gstlal_ilwdify bayestar_input/${ID}/${ID}_event.xml.gz
+DIR=$1
+ENDTIME=$2
+SID=$3
+ligolw_no_ilwdchar lloid_files/${DIR}/*CBC_AllSky-*${ENDTIME}*.xml
+FILE=$(gstlal_inspiral_best_coinc_file lloid_files/${DIR}/*CBC_AllSky-*${ENDTIME}*.xml)
+mkdir -p bayestar_input/${DIR}/
+gstlal_ligolw_add_without_reassign ${FILE} inj_files/${DIR}/*${SID}*_inj.xml.gz --output bayestar_input/${DIR}/${SID}_${ENDTIME}_event.xml.gz
+lalapps_inspinjfind --time-window 0.9 bayestar_input/${DIR}/${SID}_${ENDTIME}_event.xml.gz
+gstlal_ilwdify bayestar_input/${DIR}/${SID}_${ENDTIME}_event.xml.gz
+rm lloid_files/${DIR}/${ENDTIME}_not_used.xml.gz
+rm lloid_files/${DIR}/${ENDTIME}_not_used_2.xml.gz
 """)
 f.close()
 
-for job_id, (bankid, process_id, sid) in enumerate(s_row, start=1):
+for job_id, (templateid, bankid, process_id, sid) in enumerate(s_row, start=1):
 	# FIXME Need to add option for dist stats output
 	job_dict = {}
 	ht_gate = []
@@ -400,34 +425,55 @@ for job_id, (bankid, process_id, sid) in enumerate(s_row, start=1):
 	this_input_dict["ht-gate-threshold"] = ht_gate[ht_gate_id]
 
 	# FIX some stuff
-	fixrelpath(this_input_dict, ("reference-psd", "frame-cache", "time-slide-file", "veto-segments-file", "frame-segments-file"))
+	if not options.general_dependencies_directory:
+		fixrelpath(this_input_dict, ("reference-psd", "frame-cache", "time-slide-file", "veto-segments-file", "frame-segments-file"), analysis_dir)
+	else:
+		fixrelpath(this_input_dict, ("reference-psd", "frame-cache", "time-slide-file", "veto-segments-file", "frame-segments-file"), options.general_dependencies_directory)
 
 	# make a custom injection file
 	if options.injection_database:
-		inj_file_name = "inj_files/%d_%d_%d_inj.xml.gz" % (job_id, bankid, process_id)
-		new_inj_file(s_row[(bankid, process_id, sid)], inj_file_name)
+		coal_time = s_row[(templateid, bankid, process_id, sid)].geocent_end_time
+		injdir = "inj_files/%s" %(str(coal_time)[:5])
+		try:
+			os.mkdir(injdir)
+		except OSError:
+			pass
+		inj_file_name = "%s/%d_%d_%d_inj.xml.gz" % (injdir, sid, bankid, coal_time)
+		new_inj_file(s_row[(templateid, bankid, process_id, sid)], inj_file_name)
 		this_input_dict["injections"] = inj_file_name
+		this_opts_dict["rank-template-id"] = int(templateid)
 	else:
-		coal_time = s_row[(bankid, process_id, sid)].end_time
-		this_input_dict["reconstruction-segment"] = "%d:%d" % (coal_time - 3, coal_time + 3)
+		coal_time = s_row[(templateid, bankid, process_id, sid)].end_time
+		this_opts_dict["reconstruction-segment"] = "%d:%d" % (coal_time - 3, coal_time + 3)
+		this_opts_dict["rank-template-id"] = int(templateid)
 	# FIXME hacks for the svd
 	instruments = [x.split("=")[0] for x in this_opts_dict["channel-name"]]
-	banks = ["%s:%s" % (ifo, glob.glob("%s/gstlal_svd_bank/%s-%04d_SVD*" % (analysis_dir, ifo, bankid))[0]) for ifo in instruments]
+	if not options.general_dependencies_directory:
+		banks = ["%s:%s" % (ifo, glob.glob("%s/gstlal_svd_bank/%s-%04d_SVD*" % (analysis_dir, ifo, bankid))[0]) for ifo in instruments]
+	else:
+		banks = ["%s:%s" % (ifo, glob.glob("%s/gstlal_svd_bank/%s-%04d_SVD*" % (options.general_dependencies_directory, ifo, bankid))[0]) for ifo in instruments]
 	this_input_dict["svd-bank"] = ",".join(banks)
 
 	# FIXME don't hardcode H1L1V1
-	ranking_stat_pdf = glob.glob("%s/gstlal_inspiral_marginalize_likelihood/H1L1V1-%04d_MARG_DIST_STATS*" % (analysis_dir, bankid))[0]
+	if not options.ranking_stat_dir:
+		ranking_stat_pdf = glob.glob("%s/gstlal_inspiral_marginalize_likelihood/H1L1V1-%04d_MARG_DIST_STATS*" % (analysis_dir, bankid))[0]
+	else:
+		ranking_stat_pdf = glob.glob("%s/gstlal_inspiral_marginalize_likelihood/H1L1V1-%04d_MARG_DIST_STATS*" % (options.ranking_stat_dir, bankid))[0]
 	this_input_dict["ranking-stat-input"] = ranking_stat_pdf
 
 	# just name the output the same as the input
-	outdir = "lloid_files/%d_%d_%d" % (job_id, bankid, process_id)
+	outdir = "lloid_files/%s" % (str(coal_time)[:5])
 	try:
 		os.mkdir(outdir)
 	except OSError:
 		pass
 
-	output_file_name = "%s/%d_%d_%d_lloid.xml.gz" % (outdir, job_id, bankid, process_id)
+	output_file_name = "%s/%d_lloid.xml.gz" % (outdir, coal_time)
+	ranking_output = "%s/%d_not_used.xml.gz" % (outdir, coal_time)
+	zerolag_ranking_output = "%s/%d_not_used_2.xml.gz" % (outdir, coal_time)
 	this_output_dict["output"] = output_file_name
+	this_output_dict["ranking-stat-output"] = ranking_output
+	this_output_dict["zerolag-rankingstat-pdf"] = zerolag_ranking_output
 
 	this_opts_dict["gracedb-service-url"] = "file://%s/%s" % (os.getcwd(), outdir)
 	
@@ -435,7 +481,7 @@ for job_id, (bankid, process_id, sid) in enumerate(s_row, start=1):
 		node = dagparts.DAGNode(gstlalInspiralInjJob, dag, parent_nodes = [], opts = this_opts_dict, input_files = this_input_dict, output_files = this_output_dict)
 		# all the remaining post processing
 		# FIXME this only works for injections
-		node = dagparts.DAGNode(bashJob, dag, parent_nodes = [node], opts = {"":["process.sh", "%d_%d_%d" % (job_id, bankid, process_id)]})
+		node = dagparts.DAGNode(bashJob, dag, parent_nodes = [node], opts = {"":["process.sh", "%s %d %d" % (str(coal_time)[:5], coal_time, sid)]})
 	if options.zerolag_database:
                 node = dagparts.DAGNode(gstlalInspiralJob, dag, parent_nodes = [], opts = this_opts_dict, input_files = this_input_dict, output_files = this_output_dict)
 
diff --git a/gstlal-inspiral/python/inspiral.py b/gstlal-inspiral/python/inspiral.py
index fffa4a902..6b7ed8267 100644
--- a/gstlal-inspiral/python/inspiral.py
+++ b/gstlal-inspiral/python/inspiral.py
@@ -474,7 +474,7 @@ class FakeGracedbClient(object):
 	def __init__(self, service_url):
 		# Assumes that service url is a directory to write files to
 		self.path = urlparse.urlparse(service_url).path
-	def createEvent(self, group, pipeline, filename, filecontents, search):
+	def createEvent(self, group, pipeline, filename, filecontents, search, label):
 		with open(os.path.join(self.path, filename), "w") as f:
 			f.write(filecontents)
 		return FakeGracedbResp()
diff --git a/gstlal-inspiral/python/lloidhandler.py b/gstlal-inspiral/python/lloidhandler.py
index 285c10a8a..45625b261 100644
--- a/gstlal-inspiral/python/lloidhandler.py
+++ b/gstlal-inspiral/python/lloidhandler.py
@@ -647,7 +647,7 @@ class Handler(simplehandler.Handler):
 	dumps of segment information, trigger files and background
 	distribution statistics.
 	"""
-	def __init__(self, mainloop, pipeline, coincs_document, rankingstat, horizon_distance_func, gracedbwrapper, zerolag_rankingstatpdf_url = None, rankingstatpdf_url = None, ranking_stat_output_url = None, ranking_stat_input_url = None, likelihood_snapshot_interval = None, sngls_snr_threshold = None, tag = "", kafka_server = "10.14.0.112:9092", cluster = False, cap_singles = False, FAR_trialsfactor = 1.0, activation_counts = None, template_id_time_map = None, verbose = False):
+	def __init__(self, mainloop, pipeline, coincs_document, rankingstat, horizon_distance_func, gracedbwrapper, zerolag_rankingstatpdf_url = None, rankingstatpdf_url = None, ranking_stat_output_url = None, ranking_stat_input_url = None, likelihood_snapshot_interval = None, sngls_snr_threshold = None, tag = "", kafka_server = "10.14.0.112:9092", cluster = False, cap_singles = False, FAR_trialsfactor = 1.0, activation_counts = None, template_id_time_map = None, template_id = None, verbose = False):
 		"""!
 		@param mainloop The main application's event loop
 		@param pipeline The gstreamer pipeline that is being
@@ -674,6 +674,7 @@ class Handler(simplehandler.Handler):
 		self.cluster = cluster
 		self.cap_singles = cap_singles
 		self.FAR_trialsfactor = FAR_trialsfactor
+		self.template_id = template_id
 		self.activation_counts = activation_counts
 		self.template_id_time_map = template_id_time_map
 
@@ -746,6 +747,8 @@ class Handler(simplehandler.Handler):
 		self.ranking_stat_input_url = ranking_stat_input_url
 		self.ranking_stat_output_url = ranking_stat_output_url
 		self.rankingstat = rankingstat
+		self.rankingstat_copy = rankingstat.copy()
+		self.rankingstat_copy.finish()
 
 		#
 		# if we have been supplied with external ranking statistic
@@ -755,7 +758,7 @@ class Handler(simplehandler.Handler):
 
 		if self.ranking_stat_input_url is not None:
 			if self.rankingstat.is_healthy(self.verbose):
-				self.stream_thinca.ln_lr_from_triggers = far.OnlineFrankensteinRankingStat(self.rankingstat, self.rankingstat).finish().ln_lr_from_triggers
+				self.stream_thinca.ln_lr_from_triggers = self.rankingstat_copy.ln_lr_from_triggers
 				if self.verbose:
 					print >>sys.stderr, "ranking statistic assignment ENABLED"
 			else:
@@ -1073,7 +1076,7 @@ class Handler(simplehandler.Handler):
 				# update streamthinca's ranking statistic
 				# data
 				if self.rankingstat.is_healthy(self.verbose):
-					self.stream_thinca.ln_lr_from_triggers = far.OnlineFrankensteinRankingStat(self.rankingstat, self.rankingstat).finish().ln_lr_from_triggers
+					self.stream_thinca.ln_lr_from_triggers = self.rankingstat_copy.ln_lr_from_triggers
 					if self.verbose:
 						print >>sys.stderr, "ranking statistic assignment ENABLED"
 				else:
@@ -1132,7 +1135,7 @@ class Handler(simplehandler.Handler):
 				if not self.stream_thinca.push(instrument, [event for event in events if event.ifo == instrument], buf_timestamp):
 					continue
 
-				flushed_sngls = self.stream_thinca.pull(self.rankingstat, fapfar = self.fapfar, zerolag_rankingstatpdf = self.zerolag_rankingstatpdf, coinc_sieve = self.rankingstat.fast_path_cut_from_triggers, cluster = self.cluster, cap_singles = self.cap_singles, FAR_trialsfactor = self.FAR_trialsfactor,template_id_time_map = self.template_id_time_map)
+				flushed_sngls = self.stream_thinca.pull(self.rankingstat, fapfar = self.fapfar, zerolag_rankingstatpdf = self.zerolag_rankingstatpdf, coinc_sieve = self.rankingstat.fast_path_cut_from_triggers, cluster = self.cluster, cap_singles = self.cap_singles, FAR_trialsfactor = self.FAR_trialsfactor, template_id = self.template_id, self.template_id_time_map)
 				self.coincs_document.commit()
 
 				# do GraceDB alerts and update eye candy
@@ -1311,6 +1314,7 @@ class Handler(simplehandler.Handler):
 					del history[:endtime]
 		else:
 			rankingstat = self.rankingstat
+		rankingstat = self.rankingstat
 		far.gen_likelihood_control_doc(xmldoc, rankingstat, self.rankingstatpdf)
 		ligolw_process.set_process_end_time(process)
 		return xmldoc
@@ -1356,7 +1360,7 @@ class Handler(simplehandler.Handler):
 		# whatever triggers remain in the queues, and processes
 		# them
 
-		flushed_sngls = self.stream_thinca.pull(self.rankingstat, fapfar = self.fapfar, zerolag_rankingstatpdf = self.zerolag_rankingstatpdf, coinc_sieve = self.rankingstat.fast_path_cut_from_triggers, flush = True, cluster = self.cluster, cap_singles = self.cap_singles, FAR_trialsfactor = self.FAR_trialsfactor, template_id_time_map = self.template_id_time_map)
+		flushed_sngls = self.stream_thinca.pull(self.rankingstat, fapfar = self.fapfar, zerolag_rankingstatpdf = self.zerolag_rankingstatpdf, coinc_sieve = self.rankingstat.fast_path_cut_from_triggers, flush = True, cluster = self.cluster, cap_singles = self.cap_singles, FAR_trialsfactor = self.FAR_trialsfactor, template_id = self.template_id, template_id_time_map = self.template_id_time_map)
 		self.coincs_document.commit()
 
 		# do GraceDB alerts
diff --git a/gstlal-inspiral/python/streamthinca.py b/gstlal-inspiral/python/streamthinca.py
index 56de84309..f97f7d916 100644
--- a/gstlal-inspiral/python/streamthinca.py
+++ b/gstlal-inspiral/python/streamthinca.py
@@ -257,7 +257,7 @@ class StreamThinca(object):
 		return self.time_slide_graph.push(instrument, events, t_complete)
 
 
-	def pull(self, rankingstat, fapfar = None, zerolag_rankingstatpdf = None, coinc_sieve = None, flush = False, cluster = False, cap_singles = False, FAR_trialsfactor = 1.0, template_id_time_map = None):
+	def pull(self, rankingstat, fapfar = None, zerolag_rankingstatpdf = None, coinc_sieve = None, flush = False, cluster = False, cap_singles = False, FAR_trialsfactor = 1.0, template_id = None, template_id_time_map = None):
 		# NOTE:  rankingstat is not used to compute the ranking
 		# statistic, it supplies the detector livetime segment
 		# lists to determine which triggers are eligible for
@@ -310,7 +310,7 @@ class StreamThinca(object):
 
 			# finally, append coinc to tables
 
-			if cluster:
+			if cluster and (template_id is None or int(events[0].Gamma0) == template_id):
 				max_last_coinc_snr.setdefault(node, None)
 				if max_last_coinc_snr[node] is None or coinc_inspiral.snr > max_last_coinc_snr[node][3].snr:
 					max_last_coinc_snr[node] = (events, coinc, coincmaps, coinc_inspiral)
-- 
2.16.5

