gstlal_ll_dq: remove nxydump layer prior to appsink

gstlal-ugly/bin/gstlal_ll_dq

@@ -82,26 +82,25 @@ class NoiseTracker(object):
 
 	def on_buffer(self, buf):
 		if self.last_reduce_time is None:
-			self.last_reduce_time = int(round(buf.t0, -2))
-		if buf.data is None:
-			return
-
+			self.last_reduce_time = round(int(buf.t0), -2)
 		logging.debug(f"found buffer at t = {buf.t0}")
+
 		# First noise
-		ix = numpy.argmax(buf.data, axis=0)[1]
-		self.timedeq.append(buf.t0)
-		self.datadeq['noise'].append(buf.data[ix,1])
+		ix = numpy.argmax(buf.data[0])
+		self.timedeq.append(int(buf.t0))
+		self.datadeq['noise'].append(buf.data[0,ix])
 
-		# Then range
-		self.datadeq['range_history'].append(self.horizon_distance_func(self.psd, 8)[0] / 2.25)
+		if self.psd:
+			# Then range
+			self.datadeq['range_history'].append(self.horizon_distance_func(self.psd, 8)[0] / 2.25)
 
-		# The PSD
-		psd_freq = numpy.arange(self.psd.data.length / 4) * self.psd.deltaF * 4
-		psd_data = signal.decimate(self.psd.data.data[:], 4, ftype='fir', zero_phase=False)[:-1]**.5
+			# The PSD
+			psd_freq = numpy.arange(self.psd.data.length / 4) * self.psd.deltaF * 4
+			psd_data = signal.decimate(self.psd.data.data[:], 4, ftype='fir', zero_phase=False)[:-1]**.5
 
 		# Only reduce every 100s
 		if (buf.t0 - self.last_reduce_time) >= 100:
-			self.last_reduce_time = int(round(buf.t0, -2))
+			self.last_reduce_time = round(int(buf.t0), -2)
 			logging.info("reducing data and writing PSD snapshot for %d @ %d" % (buf.t0, int(utils.gps_now())))
 
 			data = {route: {self.instrument: {'time': list(self.timedeq), 'fields': {'data': list(self.datadeq[route])}}} for route in self.routes}
@@ -119,8 +118,8 @@ class NoiseTracker(object):
 			# NOTE: The PSD is special, we just record it. No min/median/max
 			thisdir = os.path.join(self.out_path, core.gps_to_leaf_directory(buf.t0))
 			core.makedir(thisdir)
-			psd_name = "%s-PSD-%d-100.hdf5" % (self.instrument, int(round(buf.t0, -2)))
-			self.to_hdf5(os.path.join(thisdir, psd_name), {"freq": psd_freq, "asd": psd_data, "time": numpy.array([buf.t0])})
+			psd_name = "%s-PSD-%d-100.hdf5" % (self.instrument, round(int(buf.t0), -2))
+			self.to_hdf5(os.path.join(thisdir, psd_name), {"freq": psd_freq, "asd": psd_data, "time": numpy.array([int(buf.t0)])})
 
 	def to_hdf5(self, path, datadict):
 		tmppath = "/dev/shm/%s" % path.replace("/","_") + ".tmp"
@@ -177,8 +176,7 @@ if __name__ == '__main__':
 		.whiten(fft_length=options.psd_fft_length, expand_gaps=True) \
 		.queue() \
 		.reblock() \
-		.nxydump() \
-		.sink(tracker.on_buffer)
+		.bufsink(tracker.on_buffer)
 
 	#
 	# process segment